
#include "fmacros.h"
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <errno.h>
#include <ctype.h>

#include "rediscluster.h"
#include "hiutil.h"
#include "adlist.h"
#include "hiarray.h"
#include "hiredis.h"
#include "command.h"
#include "dict.c"

#ifdef _MSC_VER
#include <WinSock2.h>
#endif

#define REDIS_COMMAND_CLUSTER_NODES "CLUSTER NODES"
#define REDIS_COMMAND_CLUSTER_SLOTS "CLUSTER SLOTS"

#define REDIS_COMMAND_ASKING "ASKING"
#define REDIS_COMMAND_PING "PING"

#define REDIS_PROTOCOL_ASKING "*1\r\n$6\r\nASKING\r\n"

#define IP_PORT_SEPARATOR ":"

#define CLUSTER_ADDRESS_SEPARATOR ","

#define CLUSTER_DEFAULT_MAX_REDIRECT_COUNT 5



static void __redisClusterSetError(redisClusterContext *cc, int type, const char *str);

static int32_t _redisClusterAuth(redisClusterContext *cc, redisContext *c)
{
	if (cc->password[0] == '\0')
		return REDIS_OK;

	int32_t result = REDIS_OK;
	redisReply* reply = redisCommand(c, "AUTH %s", cc->password);
	if (!reply) {
		if (c->err == REDIS_ERR_TIMEOUT) {
			__redisClusterSetError(cc, c->err,
				"Command(AUTH) reply error(socket timeout)");
		}
		else {
			__redisClusterSetError(cc, REDIS_ERR_OTHER,
				"Command(AUTH) reply error(NULL).");
		}
		result = REDIS_ERR;
	}
	else {

		if (reply->type == REDIS_REPLY_ERROR) {
			__redisClusterSetError(cc, REDIS_ERR_OTHER,
				reply->str);
			result = REDIS_ERR;
		}
		freeReplyObject(reply);
	}

	return result;
}


typedef struct cluster_async_data
{
	redisClusterAsyncContext *acc;
	struct cmd *command;
	redisClusterCallbackFn *callback;
	int retry_count;
	void *privdata;
}cluster_async_data;

typedef enum CLUSTER_ERR_TYPE {
	CLUSTER_NOT_ERR = 0,
	CLUSTER_ERR_MOVED,
	CLUSTER_ERR_ASK,
	CLUSTER_ERR_TRYAGAIN,
	CLUSTER_ERR_CROSSSLOT,
	CLUSTER_ERR_CLUSTERDOWN,
	CLUSTER_ERR_SENTINEL
}CLUSTER_ERR_TYPE;

static void cluster_node_deinit(cluster_node *node);
static void cluster_slot_destroy(cluster_slot *slot);
static void cluster_open_slot_destroy(copen_slot *oslot);

void listClusterNodeDestructor(void *val)
{
	cluster_node_deinit(val);

	hi_free(val);
}

void listClusterSlotDestructor(void *val)
{
	cluster_slot_destroy(val);
}

unsigned int dictSdsHash(const void *key) {
	return dictGenHashFunction((unsigned char*)key, sdslen((char*)key));
}

int dictSdsKeyCompare(void *privdata, const void *key1,
	const void *key2)
{
	int l1, l2;
	DICT_NOTUSED(privdata);

	l1 = sdslen((sds)key1);
	l2 = sdslen((sds)key2);
	if (l1 != l2) return 0;
	return memcmp(key1, key2, l1) == 0;
}

void dictSdsDestructor(void *privdata, void *val)
{
	DICT_NOTUSED(privdata);

	sdsfree(val);
}

void dictClusterNodeDestructor(void *privdata, void *val)
{
	DICT_NOTUSED(privdata);

	cluster_node_deinit(val);

	hi_free(val);
}

/* Cluster nodes hash table, mapping nodes
 * name(437c719f50dc9d0745032f3b280ce7ecc40792ac)
 * or addresses(1.2.3.4:6379) to clusterNode structures.
 * Those nodes need destroy.
 */
dictType clusterNodesDictType = {
	dictSdsHash,                /* hash function */
	NULL,                       /* key dup */
	NULL,                       /* val dup */
	dictSdsKeyCompare,          /* key compare */
	dictSdsDestructor,          /* key destructor */
	dictClusterNodeDestructor   /* val destructor */
};

/* Cluster nodes hash table, mapping nodes
 * name(437c719f50dc9d0745032f3b280ce7ecc40792ac)
 * or addresses(1.2.3.4:6379) to clusterNode structures.
 * Those nodes do not need destroy.
 */
dictType clusterNodesRefDictType = {
	dictSdsHash,                /* hash function */
	NULL,                       /* key dup */
	NULL,                       /* val dup */
	dictSdsKeyCompare,          /* key compare */
	dictSdsDestructor,          /* key destructor */
	NULL                        /* val destructor */
};


void listCommandFree(void *command)
{
	struct cmd *cmd = command;
	command_destroy(cmd);
}

/* Defined in hiredis.c */
void __redisSetError(redisContext *c, int type, const char *str);

/* Forward declaration of function in hiredis.c */
int __redisAppendCommand(redisContext *c, const char *cmd, size_t len);

/* Helper function for the redisClusterCommand* family of functions.
 *
 * Write a formatted command to the output buffer. If the given context is
 * blocking, immediately read the reply into the "reply" pointer. When the
 * context is non-blocking, the "reply" pointer will not be used and the
 * command is simply appended to the write buffer.
 *
 * Returns the reply when a reply was succesfully retrieved. Returns NULL
 * otherwise. When NULL is returned in a blocking context, the error field
 * in the context will be set.
 */
static void *__redisBlockForReply(redisContext *c) {
	void *reply;

	if (c->flags & REDIS_BLOCK) {
		if (redisGetReply(c, &reply) != REDIS_OK)
			return NULL;
		return reply;
	}
	return NULL;
}


/* -----------------------------------------------------------------------------
 * Key space handling
 * -------------------------------------------------------------------------- */

 /* We have 16384 hash slots. The hash slot of a given key is obtained
  * as the least significant 14 bits of the crc16 of the key.
  *
  * However if the key contains the {...} pattern, only the part between
  * { and } is hashed. This may be useful in the future to force certain
  * keys to be in the same node (assuming no resharding is in progress). */
static unsigned int keyHashSlot(char *key, int keylen) {
	int s, e; /* start-end indexes of { and } */

	for (s = 0; s < keylen; s++)
		if (key[s] == '{') break;

	/* No '{' ? Hash the whole key. This is the base case. */
	if (s == keylen) return crc16(key, keylen) & 0x3FFF;

	/* '{' found? Check if we have the corresponding '}'. */
	for (e = s + 1; e < keylen; e++)
		if (key[e] == '}') break;

	/* No '}' or nothing betweeen {} ? Hash the whole key. */
	if (e == keylen || e == s + 1) return crc16(key, keylen) & 0x3FFF;

	/* If we are here there is both a { and a } on its right. Hash
	 * what is in the middle between { and }. */
	return crc16(key + s + 1, e - s - 1) & 0x3FFF;
}

static void __redisClusterSetError(redisClusterContext *cc, int type, const char *str) {

	if (cc == NULL) {
		return;
	}

	cc->err = type;
	if (str != NULL) {
		strncpy(cc->errstr, str, sizeof(cc->errstr));
	}
	else {
		/* Only REDIS_ERR_IO may lack a description! */
		assert(type == REDIS_ERR_IO);
		__redis_strerror_r(errno, cc->errstr, sizeof(cc->errstr));
	}
}

static int cluster_reply_error_type(redisReply *reply)
{

	if (reply == NULL)
	{
		return REDIS_ERR;
	}

	if (reply->type == REDIS_REPLY_ERROR)
	{
		if ((int)strlen(REDIS_ERROR_MOVED) < reply->len &&
			strncmp(reply->str, REDIS_ERROR_MOVED, strlen(REDIS_ERROR_MOVED)) == 0)
		{
			return CLUSTER_ERR_MOVED;
		}
		else if ((int)strlen(REDIS_ERROR_ASK) < reply->len &&
			strncmp(reply->str, REDIS_ERROR_ASK, strlen(REDIS_ERROR_ASK)) == 0)
		{
			return CLUSTER_ERR_ASK;
		}
		else if ((int)strlen(REDIS_ERROR_TRYAGAIN) < reply->len &&
			strncmp(reply->str, REDIS_ERROR_TRYAGAIN, strlen(REDIS_ERROR_TRYAGAIN)) == 0)
		{
			return CLUSTER_ERR_TRYAGAIN;
		}
		else if ((int)strlen(REDIS_ERROR_CROSSSLOT) < reply->len &&
			strncmp(reply->str, REDIS_ERROR_CROSSSLOT, strlen(REDIS_ERROR_CROSSSLOT)) == 0)
		{
			return CLUSTER_ERR_CROSSSLOT;
		}
		else if ((int)strlen(REDIS_ERROR_CLUSTERDOWN) < reply->len &&
			strncmp(reply->str, REDIS_ERROR_CLUSTERDOWN, strlen(REDIS_ERROR_CLUSTERDOWN)) == 0)
		{
			return CLUSTER_ERR_CLUSTERDOWN;
		}
		else
		{
			return CLUSTER_ERR_SENTINEL;
		}
	}

	return CLUSTER_NOT_ERR;
}

static int cluster_node_init(cluster_node *node)
{
	if (node == NULL) {
		return REDIS_ERR;
	}

	node->name = NULL;
	node->addr = NULL;
	node->host = NULL;
	node->port = 0;
	node->role = REDIS_ROLE_NULL;
	node->myself = 0;
	node->slaves = NULL;
	node->con = NULL;
	node->acon = NULL;
	node->slots = NULL;
	node->failure_count = 0;
	node->data = NULL;
	node->migrating = NULL;
	node->importing = NULL;

	return REDIS_OK;
}

static void cluster_node_deinit(cluster_node *node)
{
	copen_slot **oslot;

	if (node == NULL)
	{
		return;
	}

	sdsfree(node->name);
	sdsfree(node->addr);
	sdsfree(node->host);
	node->port = 0;
	node->role = REDIS_ROLE_NULL;
	node->myself = 0;

	if (node->con != NULL)
	{
		redisFree(node->con);
	}

	if (node->acon != NULL)
	{
		redisAsyncFree(node->acon);
	}

	if (node->slots != NULL)
	{
		listRelease(node->slots);
	}

	if (node->slaves != NULL)
	{
		listRelease(node->slaves);
	}

	if (node->migrating)
	{
		while (hiarray_n(node->migrating))
		{
			oslot = hiarray_pop(node->migrating);
			cluster_open_slot_destroy(*oslot);
		}

		hiarray_destroy(node->migrating);
		node->migrating = NULL;
	}

	if (node->importing)
	{
		while (hiarray_n(node->importing))
		{
			oslot = hiarray_pop(node->importing);
			cluster_open_slot_destroy(*oslot);
		}

		hiarray_destroy(node->importing);
		node->importing = NULL;
	}
}

static int cluster_slot_init(cluster_slot *slot, cluster_node *node)
{
	slot->start = 0;
	slot->end = 0;
	slot->node = node;

	return REDIS_OK;
}

static cluster_slot *cluster_slot_create(cluster_node *node)
{
	cluster_slot *slot;

	slot = hi_alloc(sizeof(*slot));
	if (slot == NULL) {
		return NULL;
	}

	cluster_slot_init(slot, node);

	if (node != NULL) {
		ASSERT(node->role == REDIS_ROLE_MASTER);
		if (node->slots == NULL) {
			node->slots = listCreate();
			if (node->slots == NULL)
			{
				cluster_slot_destroy(slot);
				return NULL;
			}

			node->slots->free = listClusterSlotDestructor;
		}

		listAddNodeTail(node->slots, slot);
	}

	return slot;
}

static int cluster_slot_ref_node(cluster_slot * slot, cluster_node *node)
{
	if (slot == NULL || node == NULL) {
		return REDIS_ERR;
	}


	if (node->role != REDIS_ROLE_MASTER) {
		return REDIS_ERR;
	}

	if (node->slots == NULL) {
		node->slots = listCreate();
		if (node->slots == NULL)
		{
			return REDIS_ERR;
		}

		node->slots->free = listClusterSlotDestructor;
	}

	listAddNodeTail(node->slots, slot);
	slot->node = node;

	return REDIS_OK;
}

static void cluster_slot_destroy(cluster_slot *slot)
{
	slot->start = 0;
	slot->end = 0;
	slot->node = NULL;

	hi_free(slot);
}

static copen_slot *cluster_open_slot_create(uint32_t slot_num, int migrate,
	sds remote_name, cluster_node *node)
{
	copen_slot *oslot;

	oslot = hi_alloc(sizeof(*oslot));
	if (oslot == NULL) {
		return NULL;
	}

	oslot->slot_num = 0;
	oslot->migrate = 0;
	oslot->node = NULL;
	oslot->remote_name = NULL;

	oslot->slot_num = slot_num;
	oslot->migrate = migrate;
	oslot->node = node;
	oslot->remote_name = sdsdup(remote_name);

	return oslot;
}

static void cluster_open_slot_destroy(copen_slot *oslot)
{
	oslot->slot_num = 0;
	oslot->migrate = 0;
	oslot->node = NULL;

	if (oslot->remote_name != NULL) {
		sdsfree(oslot->remote_name);
		oslot->remote_name = NULL;
	}

	hi_free(oslot);
}

/**
  * Return a new node with the "cluster slots" command reply.
  */
static cluster_node *node_get_with_slots(
	redisClusterContext *cc, redisReply *host_elem,
	redisReply *port_elem, uint8_t role)
{
	cluster_node *node = NULL;

	if (host_elem == NULL || port_elem == NULL) {
		return NULL;
	}

	if (host_elem->type != REDIS_REPLY_STRING ||
		host_elem->len <= 0) {
		__redisClusterSetError(cc, REDIS_ERR_OTHER,
			"Command(cluster slots) reply error: "
			"node ip is not string.");
		goto error;
	}

	if (port_elem->type != REDIS_REPLY_INTEGER ||
		port_elem->integer <= 0) {
		__redisClusterSetError(cc, REDIS_ERR_OTHER,
			"Command(cluster slots) reply error: "
			"node port is not integer.");
		goto error;
	}

	if (!hi_valid_port((int)port_elem->integer)) {
		__redisClusterSetError(cc, REDIS_ERR_OTHER,
			"Command(cluster slots) reply error: "
			"node port is not valid.");
		goto error;
	}

	node = hi_alloc(sizeof(cluster_node));
	if (node == NULL) {
		__redisClusterSetError(cc,
			REDIS_ERR_OOM, "Out of memory");
		goto error;
	}

	cluster_node_init(node);

	if (role == REDIS_ROLE_MASTER) {
		node->slots = listCreate();
		if (node->slots == NULL) {
			hi_free(node);
			__redisClusterSetError(cc, REDIS_ERR_OTHER,
				"slots for node listCreate error");
			goto error;
		}

		node->slots->free = listClusterSlotDestructor;
	}

	node->name = NULL;
	node->addr = sdsnewlen(host_elem->str, host_elem->len);
	node->addr = sdscatfmt(node->addr, ":%i", port_elem->integer);

	node->host = sdsnewlen(host_elem->str, host_elem->len);
	node->port = (int)port_elem->integer;
	node->role = role;

	return node;

error:

	if (node != NULL) {
		hi_free(node);
	}

	return NULL;
}

/**
  * Return a new node with the "cluster nodes" command reply.
  */
static cluster_node *node_get_with_nodes(
	redisClusterContext *cc,
	sds *node_infos, int info_count, uint8_t role)
{
	sds *ip_port = NULL;
	int count_ip_port = 0;
	cluster_node *node;

	if (info_count < 8)
	{
		return NULL;
	}

	node = hi_alloc(sizeof(cluster_node));
	if (node == NULL)
	{
		__redisClusterSetError(cc,
			REDIS_ERR_OOM, "Out of memory");
		goto error;
	}

	cluster_node_init(node);

	if (role == REDIS_ROLE_MASTER)
	{
		node->slots = listCreate();
		if (node->slots == NULL)
		{
			hi_free(node);
			__redisClusterSetError(cc, REDIS_ERR_OTHER,
				"slots for node listCreate error");
			goto error;
		}

		node->slots->free = listClusterSlotDestructor;
	}

	node->name = node_infos[0];
	node->addr = node_infos[1];

	ip_port = sdssplitlen(node_infos[1], sdslen(node_infos[1]),
		IP_PORT_SEPARATOR, strlen(IP_PORT_SEPARATOR), &count_ip_port);
	if (ip_port == NULL || count_ip_port != 2)
	{
		__redisClusterSetError(cc, REDIS_ERR_OTHER,
			"split ip port error");
		goto error;
	}
	node->host = ip_port[0];
	//node->port = hi_atoi(ip_port[1], sdslen(ip_port[1]));
	node->port = atoi(ip_port[1]);
	node->role = role;

	sdsfree(ip_port[1]);
	free(ip_port);

	node_infos[0] = NULL;
	node_infos[1] = NULL;

	return node;

error:
	if (ip_port != NULL)
	{
		sdsfreesplitres(ip_port, count_ip_port);
	}

	if (node != NULL)
	{
		hi_free(node);
	}

	return NULL;
}

static void cluster_nodes_swap_ctx(dict *nodes_f, dict *nodes_t)
{
	dictIterator *di;
	dictEntry *de_f, *de_t;
	cluster_node *node_f, *node_t;
	redisContext *c;
	redisAsyncContext *ac;

	if (nodes_f == NULL || nodes_t == NULL) {
		return;
	}

	di = dictGetIterator(nodes_t);
	while ((de_t = dictNext(di)) != NULL) {
		node_t = dictGetEntryVal(de_t);
		if (node_t == NULL) {
			continue;
		}

		de_f = dictFind(nodes_f, node_t->addr);
		if (de_f == NULL) {
			continue;
		}

		node_f = dictGetEntryVal(de_f);
		if (node_f->con != NULL) {
			c = node_f->con;
			node_f->con = node_t->con;
			node_t->con = c;
		}

		if (node_f->acon != NULL) {
			ac = node_f->acon;
			node_f->acon = node_t->acon;
			node_t->acon = ac;

			node_t->acon->data = node_t;
			if (node_f->acon)
				node_f->acon->data = node_f;
		}
	}

	dictReleaseIterator(di);

}

static int
cluster_slot_start_cmp(const void *t1, const void *t2)
{
	const cluster_slot **s1 = (const cluster_slot **)t1, **s2 = (const cluster_slot **)t2;

	return (*s1)->start > (*s2)->start ? 1 : -1;
}

static int
cluster_master_slave_mapping_with_name(redisClusterContext *cc,
	dict **nodes, cluster_node *node, sds master_name)
{
	int ret;
	dictEntry *di;
	cluster_node *node_old;
	listNode *lnode;

	if (node == NULL || master_name == NULL)
	{
		return REDIS_ERR;
	}

	if (*nodes == NULL)
	{
		*nodes = dictCreate(
			&clusterNodesRefDictType, NULL);
	}

	di = dictFind(*nodes, master_name);
	if (di == NULL)
	{
		ret = dictAdd(*nodes,
			sdsnewlen(master_name, sdslen(master_name)), node);
		if (ret != DICT_OK)
		{
			__redisClusterSetError(cc, REDIS_ERR_OTHER,
				"the address already exists in the nodes");
			return REDIS_ERR;
		}

	}
	else
	{
		node_old = dictGetEntryVal(di);
		if (node_old == NULL)
		{
			__redisClusterSetError(cc, REDIS_ERR_OTHER,
				"dict get value null");
			return REDIS_ERR;
		}

		if (node->role == REDIS_ROLE_MASTER &&
			node_old->role == REDIS_ROLE_MASTER)
		{
			__redisClusterSetError(cc, REDIS_ERR_OTHER,
				"two masters have the same name");
			return REDIS_ERR;
		}
		else if (node->role == REDIS_ROLE_MASTER
			&& node_old->role == REDIS_ROLE_SLAVE)
		{
			if (node->slaves == NULL)
			{
				node->slaves = listCreate();
				if (node->slaves == NULL)
				{
					__redisClusterSetError(cc, REDIS_ERR_OOM,
						"Out of memory");
					return REDIS_ERR;
				}

				node->slaves->free =
					listClusterNodeDestructor;
			}

			if (node_old->slaves != NULL)
			{
				node_old->slaves->free = NULL;
				while (listLength(node_old->slaves) > 0)
				{
					lnode = listFirst(node_old->slaves);
					listAddNodeHead(node->slaves, lnode->value);
					listDelNode(node_old->slaves, lnode);
				}
				listRelease(node_old->slaves);
				node_old->slaves = NULL;
			}

			listAddNodeHead(node->slaves, node_old);

			dictSetHashVal(*nodes, di, node);
		}
		else if (node->role == REDIS_ROLE_SLAVE)
		{
			if (node_old->slaves == NULL)
			{
				node_old->slaves = listCreate();
				if (node_old->slaves == NULL)
				{
					__redisClusterSetError(cc, REDIS_ERR_OOM,
						"Out of memory");
					return REDIS_ERR;
				}

				node_old->slaves->free =
					listClusterNodeDestructor;
			}

			listAddNodeTail(node_old->slaves, node);
		}
		else
		{
			NOT_REACHED();
		}
	}

	return REDIS_OK;
}

/**
  * Parse the "cluster slots" command reply to nodes dict.
  */
dict *
parse_cluster_slots(redisClusterContext *cc,
	redisReply *reply, int flags)
{
	int ret;
	cluster_slot *slot = NULL;
	dict *nodes = NULL;
	dictEntry *den;
	redisReply *elem_slots;
	redisReply *elem_slots_begin, *elem_slots_end;
	redisReply *elem_nodes;
	redisReply *elem_ip, *elem_port;
	cluster_node *master = NULL, *slave;
	sds address;
	uint32_t i, idx;

	if (reply == NULL) {
		return NULL;
	}

	nodes = dictCreate(&clusterNodesDictType, NULL);
	if (nodes == NULL) {
		__redisClusterSetError(cc, REDIS_ERR_OOM,
			"out of memory");
		goto error;
	}

	if (reply->type != REDIS_REPLY_ARRAY || reply->elements <= 0) {
		__redisClusterSetError(cc, REDIS_ERR_OTHER,
			"Command(cluster slots) reply error: "
			"reply is not an array.");
		goto error;
	}

	for (i = 0; i < reply->elements; i++) {
		elem_slots = reply->element[i];
		if (elem_slots->type != REDIS_REPLY_ARRAY ||
			elem_slots->elements < 3) {
			__redisClusterSetError(cc, REDIS_ERR_OTHER,
				"Command(cluster slots) reply error: "
				"first sub_reply is not an array.");
			goto error;
		}

		slot = cluster_slot_create(NULL);
		if (slot == NULL) {
			__redisClusterSetError(cc, REDIS_ERR_OOM,
				"Slot create failed: out of memory.");
			goto error;
		}

		//one slots region
		for (idx = 0; idx < elem_slots->elements; idx++) {
			if (idx == 0) {
				elem_slots_begin = elem_slots->element[idx];
				if (elem_slots_begin->type != REDIS_REPLY_INTEGER) {
					__redisClusterSetError(cc, REDIS_ERR_OTHER,
						"Command(cluster slots) reply error: "
						"slot begin is not an integer.");
					goto error;
				}
				slot->start = (int)(elem_slots_begin->integer);
			}
			else if (idx == 1) {
				elem_slots_end = elem_slots->element[idx];
				if (elem_slots_end->type != REDIS_REPLY_INTEGER) {
					__redisClusterSetError(cc, REDIS_ERR_OTHER,
						"Command(cluster slots) reply error: "
						"slot end is not an integer.");
					goto error;
				}

				slot->end = (int)(elem_slots_end->integer);

				if (slot->start > slot->end) {
					__redisClusterSetError(cc, REDIS_ERR_OTHER,
						"Command(cluster slots) reply error: "
						"slot begin is bigger than slot end.");
					goto error;
				}
			}
			else {
				elem_nodes = elem_slots->element[idx];
				if (elem_nodes->type != REDIS_REPLY_ARRAY ||
					elem_nodes->elements != 3) {
					__redisClusterSetError(cc, REDIS_ERR_OTHER,
						"Command(cluster slots) reply error: "
						"nodes sub_reply is not an correct array.");
					goto error;
				}

				elem_ip = elem_nodes->element[0];
				elem_port = elem_nodes->element[1];

				if (elem_ip == NULL || elem_port == NULL ||
					elem_ip->type != REDIS_REPLY_STRING ||
					elem_port->type != REDIS_REPLY_INTEGER) {
					__redisClusterSetError(cc, REDIS_ERR_OTHER,
						"Command(cluster slots) reply error: "
						"master ip or port is not correct.");
					goto error;
				}

				//this is master.
				if (idx == 2) {
					address = sdsnewlen(elem_ip->str, elem_ip->len);
					address = sdscatfmt(address, ":%i", elem_port->integer);

					den = dictFind(nodes, address);
					//master already exits, break to the next slots region.
					if (den != NULL) {
						sdsfree(address);

						master = dictGetEntryVal(den);
						ret = cluster_slot_ref_node(slot, master);
						if (ret != REDIS_OK) {
							__redisClusterSetError(cc, REDIS_ERR_OOM,
								"Slot ref node failed: out of memory.");
							goto error;
						}

						slot = NULL;
						break;
					}

					sdsfree(address);
					master = node_get_with_slots(cc, elem_ip,
						elem_port, REDIS_ROLE_MASTER);
					if (master == NULL) {
						goto error;
					}

					ret = dictAdd(nodes,
						sdsnewlen(master->addr, sdslen(master->addr)), master);
					if (ret != DICT_OK) {
						__redisClusterSetError(cc, REDIS_ERR_OTHER,
							"The address already exists in the nodes");
						cluster_node_deinit(master);
						hi_free(master);
						goto error;
					}

					ret = cluster_slot_ref_node(slot, master);
					if (ret != REDIS_OK) {
						__redisClusterSetError(cc, REDIS_ERR_OOM,
							"Slot ref node failed: out of memory.");
						goto error;
					}

					slot = NULL;
				}
				else if (flags & HIRCLUSTER_FLAG_ADD_SLAVE) {
					slave = node_get_with_slots(cc, elem_ip,
						elem_port, REDIS_ROLE_SLAVE);
					if (slave == NULL) {
						goto error;
					}

					if (master->slaves == NULL) {
						master->slaves = listCreate();
						if (master->slaves == NULL) {
							__redisClusterSetError(cc, REDIS_ERR_OOM,
								"Out of memory");
							cluster_node_deinit(slave);
							goto error;
						}

						master->slaves->free =
							listClusterNodeDestructor;
					}

					listAddNodeTail(master->slaves, slave);
				}
			}
		}
	}

	return nodes;

error:

	if (nodes != NULL) {
		dictRelease(nodes);
	}

	if (slot != NULL) {
		cluster_slot_destroy(slot);
	}

	return NULL;
}

/**
  * Parse the "cluster nodes" command reply to nodes dict.
  */
dict *
parse_cluster_nodes(redisClusterContext *cc,
	char *str, int str_len, int flags)
{
	int ret;
	dict *nodes = NULL;
	dict *nodes_name = NULL;
	cluster_node *master, *slave;
	cluster_slot *slot;
	char *pos, *start, *end, *line_start, *line_end;
	char *role;
	int role_len;
	uint8_t myself = 0;
	int slot_start, slot_end;
	sds *part = NULL, *slot_start_end = NULL;
	int count_part = 0, count_slot_start_end = 0;
	int k;
	int len;

	nodes = dictCreate(&clusterNodesDictType, NULL);
	if (nodes == NULL) {
		__redisClusterSetError(cc, REDIS_ERR_OOM,
			"out of memory");
		goto error;
	}

	start = str;
	end = start + str_len;

	line_start = start;

	for (pos = start; pos < end; pos++) {
		if (*pos == '\n') {
			line_end = pos - 1;
			len = line_end - line_start;

			part = sdssplitlen(line_start, len + 1, " ", 1, &count_part);

			if (part == NULL || count_part < 8) {
				__redisClusterSetError(cc, REDIS_ERR_OTHER,
					"split cluster nodes error");
				goto error;
			}

			//the address string is ":0", skip this node.
			if (sdslen(part[1]) == 2 && strcmp(part[1], ":0") == 0) {
				sdsfreesplitres(part, count_part);
				count_part = 0;
				part = NULL;

				start = pos + 1;
				line_start = start;
				pos = start;

				continue;
			}

			if (sdslen(part[2]) >= 7 && memcmp(part[2], "myself,", 7) == 0) {
				role_len = sdslen(part[2]) - 7;
				role = part[2] + 7;
				myself = 1;
			}
			else {
				role_len = sdslen(part[2]);
				role = part[2];
			}

			//add master node
			if (role_len >= 6 && memcmp(role, "master", 6) == 0) {
				if (count_part < 8) {
					__redisClusterSetError(cc, REDIS_ERR_OTHER,
						"Master node parts number error: less than 8.");
					goto error;
				}

				master = node_get_with_nodes(cc,
					part, count_part, REDIS_ROLE_MASTER);
				if (master == NULL) {
					goto error;
				}

				ret = dictAdd(nodes,
					sdsnewlen(master->addr, sdslen(master->addr)), master);
				if (ret != DICT_OK) {
					__redisClusterSetError(cc, REDIS_ERR_OTHER,
						"The address already exists in the nodes");
					cluster_node_deinit(master);
					hi_free(master);
					goto error;
				}

				if (flags & HIRCLUSTER_FLAG_ADD_SLAVE) {
					ret = cluster_master_slave_mapping_with_name(cc,
						&nodes_name, master, master->name);
					if (ret != REDIS_OK) {
						cluster_node_deinit(master);
						hi_free(master);
						goto error;
					}
				}

				if (myself) master->myself = 1;

				for (k = 8; k < count_part; k++) {
					slot_start_end = sdssplitlen(part[k],
						sdslen(part[k]), "-", 1, &count_slot_start_end);

					if (slot_start_end == NULL) {
						__redisClusterSetError(cc, REDIS_ERR_OTHER,
							"split slot start end error(NULL)");
						goto error;
					}
					else if (count_slot_start_end == 1) {
						slot_start =
							hi_atoi(slot_start_end[0], sdslen(slot_start_end[0]));
						slot_end = slot_start;
					}
					else if (count_slot_start_end == 2) {
						slot_start =
							hi_atoi(slot_start_end[0], sdslen(slot_start_end[0]));;
						slot_end =
							hi_atoi(slot_start_end[1], sdslen(slot_start_end[1]));;
					}
					else {
						//add open slot for master
						if (flags & HIRCLUSTER_FLAG_ADD_OPENSLOT &&
							count_slot_start_end == 3 &&
							sdslen(slot_start_end[0]) > 1 &&
							sdslen(slot_start_end[1]) == 1 &&
							sdslen(slot_start_end[2]) > 1 &&
							slot_start_end[0][0] == '[' &&
							slot_start_end[2][sdslen(slot_start_end[2]) - 1] == ']') {

							copen_slot *oslot, **oslot_elem;

							sdsrange(slot_start_end[0], 1, -1);
							sdsrange(slot_start_end[2], 0, -2);

							if (slot_start_end[1][0] == '>') {
								oslot = cluster_open_slot_create(
									hi_atoi(slot_start_end[0],
										sdslen(slot_start_end[0])),
									1, slot_start_end[2], master);
								if (oslot == NULL) {
									__redisClusterSetError(cc, REDIS_ERR_OTHER,
										"create open slot error");
									goto error;
								}

								if (master->migrating == NULL) {
									master->migrating = hiarray_create(1, sizeof(oslot));
									if (master->migrating == NULL) {
										__redisClusterSetError(cc, REDIS_ERR_OTHER,
											"create migrating array error");
										cluster_open_slot_destroy(oslot);
										goto error;
									}
								}

								oslot_elem = hiarray_push(master->migrating);
								if (oslot_elem == NULL) {
									__redisClusterSetError(cc, REDIS_ERR_OTHER,
										"Push migrating array error: out of memory");
									cluster_open_slot_destroy(oslot);
									goto error;
								}

								*oslot_elem = oslot;
							}
							else if (slot_start_end[1][0] == '<') {
								oslot = cluster_open_slot_create(hi_atoi(slot_start_end[0],
									sdslen(slot_start_end[0])), 0, slot_start_end[2],
									master);
								if (oslot == NULL) {
									__redisClusterSetError(cc, REDIS_ERR_OTHER,
										"create open slot error");
									goto error;
								}

								if (master->importing == NULL) {
									master->importing = hiarray_create(1, sizeof(oslot));
									if (master->importing == NULL) {
										__redisClusterSetError(cc, REDIS_ERR_OTHER,
											"create migrating array error");
										cluster_open_slot_destroy(oslot);
										goto error;
									}
								}

								oslot_elem = hiarray_push(master->importing);
								if (oslot_elem == NULL) {
									__redisClusterSetError(cc, REDIS_ERR_OTHER,
										"push migrating array error: out of memory");
									cluster_open_slot_destroy(oslot);
									goto error;
								}

								*oslot_elem = oslot;
							}
						}

						slot_start = -1;
						slot_end = -1;
					}

					sdsfreesplitres(slot_start_end, count_slot_start_end);
					count_slot_start_end = 0;
					slot_start_end = NULL;

					if (slot_start < 0 || slot_end < 0 ||
						slot_start > slot_end || slot_end >= REDIS_CLUSTER_SLOTS) {
						continue;
					}

					slot = cluster_slot_create(master);
					if (slot == NULL) {
						__redisClusterSetError(cc, REDIS_ERR_OOM,
							"Out of memory");
						goto error;
					}

					slot->start = (uint32_t)slot_start;
					slot->end = (uint32_t)slot_end;
				}

			}
			//add slave node
			else if ((flags & HIRCLUSTER_FLAG_ADD_SLAVE) &&
				(role_len >= 5 && memcmp(role, "slave", 5) == 0)) {
				slave = node_get_with_nodes(cc, part,
					count_part, REDIS_ROLE_SLAVE);
				if (slave == NULL) {
					goto error;
				}

				ret = cluster_master_slave_mapping_with_name(cc,
					&nodes_name, slave, part[3]);
				if (ret != REDIS_OK) {
					cluster_node_deinit(slave);
					hi_free(slave);
					goto error;
				}

				if (myself) slave->myself = 1;
			}

			if (myself == 1) {
				myself = 0;
			}

			sdsfreesplitres(part, count_part);
			count_part = 0;
			part = NULL;

			start = pos + 1;
			line_start = start;
			pos = start;
		}
	}

	if (nodes_name != NULL) {
		dictRelease(nodes_name);
	}

	return nodes;

error:

	if (part != NULL) {
		sdsfreesplitres(part, count_part);
		count_part = 0;
		part = NULL;
	}

	if (slot_start_end != NULL) {
		sdsfreesplitres(slot_start_end, count_slot_start_end);
		count_slot_start_end = 0;
		slot_start_end = NULL;
	}

	if (nodes != NULL) {
		dictRelease(nodes);
	}

	if (nodes_name != NULL) {
		dictRelease(nodes_name);
	}

	return NULL;
}

/**
  * Update route with the "cluster nodes" or "cluster slots" command reply.
  */
static int
cluster_update_route_by_addr(redisClusterContext *cc,
	const char *ip, int port)
{
	redisContext *c = NULL;
	redisReply *reply = NULL;
	dict *nodes = NULL;
	struct hiarray *slots = NULL;
	cluster_node *master;
	cluster_slot *slot, **slot_elem;
	dictIterator *dit = NULL;
	dictEntry *den;
	listIter *lit = NULL;
	listNode *lnode;
	cluster_node *table[REDIS_CLUSTER_SLOTS];
	uint32_t j, k;

	if (cc == NULL) {
		return REDIS_ERR;
	}

	if (ip == NULL || port <= 0) {
		__redisClusterSetError(cc,
			REDIS_ERR_OTHER, "Ip or port error!");
		goto error;
	}

	if (cc->connect_timeout) {
		c = redisConnectWithTimeout(ip, port, *cc->connect_timeout);
	}
	else {
		c = redisConnect(ip, port);
	}

	if (c == NULL) {
		__redisClusterSetError(cc, REDIS_ERR_OTHER,
			"Init redis context error(return NULL)");
		goto error;
	}
	else if (c->err) {
		__redisClusterSetError(cc, c->err, c->errstr);
		goto error;
	}


	if (cc->timeout) {
		redisSetTimeout(c, *cc->timeout);
	}

	if (_redisClusterAuth(cc, c) != REDIS_OK) {
		goto error;
	}


	if (cc->flags & HIRCLUSTER_FLAG_ROUTE_USE_SLOTS) {
		reply = redisCommand(c, REDIS_COMMAND_CLUSTER_SLOTS);
		if (reply == NULL) {
			if (c->err == REDIS_ERR_TIMEOUT) {
				__redisClusterSetError(cc, c->err,
					"Command(cluster slots) reply error(socket timeout)");
			}
			else {
				__redisClusterSetError(cc, REDIS_ERR_OTHER,
					"Command(cluster slots) reply error(NULL).");
			}
			goto error;
		}
		else if (reply->type != REDIS_REPLY_ARRAY) {
			if (reply->type == REDIS_REPLY_ERROR) {
				__redisClusterSetError(cc, REDIS_ERR_OTHER,
					reply->str);
			}
			else {
				__redisClusterSetError(cc, REDIS_ERR_OTHER,
					"Command(cluster slots) reply error: type is not array.");
			}

			goto error;
		}

		nodes = parse_cluster_slots(cc, reply, cc->flags);
	}
	else {
		reply = redisCommand(c, REDIS_COMMAND_CLUSTER_NODES);
		if (reply == NULL) {
			if (c->err == REDIS_ERR_TIMEOUT) {
				__redisClusterSetError(cc, c->err,
					"Command(cluster nodes) reply error(socket timeout)");
			}
			else {
				__redisClusterSetError(cc, REDIS_ERR_OTHER,
					"Command(cluster nodes) reply error(NULL).");
			}
			goto error;
		}
		else if (reply->type != REDIS_REPLY_STRING) {
			if (reply->type == REDIS_REPLY_ERROR) {
				__redisClusterSetError(cc, REDIS_ERR_OTHER,
					reply->str);
			}
			else {
				__redisClusterSetError(cc, REDIS_ERR_OTHER,
					"Command(cluster nodes) reply error: type is not string.");
			}

			goto error;
		}

		nodes = parse_cluster_nodes(cc, reply->str, reply->len, cc->flags);
	}

	if (nodes == NULL) {
		goto error;
	}

	memset(table, 0, REDIS_CLUSTER_SLOTS * sizeof(cluster_node *));

	slots = hiarray_create(dictSize(nodes), sizeof(cluster_slot*));
	if (slots == NULL) {
		__redisClusterSetError(cc, REDIS_ERR_OTHER,
			"Slots array create failed: out of memory");
		goto error;
	}

	dit = dictGetIterator(nodes);
	if (dit == NULL) {
		__redisClusterSetError(cc, REDIS_ERR_OOM,
			"Dict get iterator failed: out of memory");
		goto error;
	}

	while ((den = dictNext(dit))) {
		master = dictGetEntryVal(den);
		if (master->role != REDIS_ROLE_MASTER) {
			__redisClusterSetError(cc, REDIS_ERR_OOM,
				"Node role must be master");
			goto error;
		}

		if (master->slots == NULL) {
			continue;
		}

		lit = listGetIterator(master->slots, AL_START_HEAD);
		if (lit == NULL) {
			__redisClusterSetError(cc, REDIS_ERR_OOM,
				"List get iterator failed: out of memory");
			goto error;
		}

		while ((lnode = listNext(lit))) {
			slot = listNodeValue(lnode);
			if (slot->start > slot->end ||
				slot->end >= REDIS_CLUSTER_SLOTS) {
				__redisClusterSetError(cc, REDIS_ERR_OTHER,
					"Slot region for node is error");
				goto error;
			}

			slot_elem = hiarray_push(slots);
			*slot_elem = slot;
		}

		listReleaseIterator(lit);
	}

	dictReleaseIterator(dit);

	hiarray_sort(slots, cluster_slot_start_cmp);
	for (j = 0; j < hiarray_n(slots); j++) {
		slot_elem = hiarray_get(slots, j);

		for (k = (*slot_elem)->start; k <= (*slot_elem)->end; k++) {
			if (table[k] != NULL) {
				__redisClusterSetError(cc, REDIS_ERR_OTHER,
					"Diffent node hold a same slot");
				goto error;
			}

			table[k] = (*slot_elem)->node;
		}
	}

	cluster_nodes_swap_ctx(cc->nodes, nodes);
	if (cc->nodes != NULL) {
		dictRelease(cc->nodes);
		cc->nodes = NULL;
	}
	cc->nodes = nodes;

	if (cc->slots != NULL)
	{
		cc->slots->nelem = 0;
		hiarray_destroy(cc->slots);
		cc->slots = NULL;
	}
	cc->slots = slots;

	memcpy(cc->table, table, REDIS_CLUSTER_SLOTS * sizeof(cluster_node *));
	cc->route_version++;

	freeReplyObject(reply);

	if (c != NULL) {
		redisFree(c);
	}

	return REDIS_OK;

error:

	if (dit != NULL) {
		dictReleaseIterator(dit);
	}

	if (lit != NULL) {
		listReleaseIterator(lit);
	}

	if (slots != NULL)
	{
		if (slots == cc->slots)
		{
			cc->slots = NULL;
		}

		slots->nelem = 0;
		hiarray_destroy(slots);
	}

	if (nodes != NULL) {
		if (nodes == cc->nodes) {
			cc->nodes = NULL;
		}

		dictRelease(nodes);
	}

	if (reply != NULL) {
		freeReplyObject(reply);
		reply = NULL;
	}

	if (c != NULL) {
		redisFree(c);
	}

	return REDIS_ERR;
}


/**
  * Update route with the "cluster nodes" command reply.
  */
static int
cluster_update_route_with_nodes_old(redisClusterContext *cc,
	const char *ip, int port)
{
	int ret;
	redisContext *c = NULL;
	redisReply *reply = NULL;
	struct hiarray *slots = NULL;
	dict *nodes = NULL;
	dict *nodes_name = NULL;
	cluster_node *master, *slave;
	cluster_slot **slot;
	char *pos, *start, *end, *line_start, *line_end;
	char *role;
	int role_len;
	uint8_t myself = 0;
	int slot_start, slot_end;
	sds *part = NULL, *slot_start_end = NULL;
	int count_part = 0, count_slot_start_end = 0;
	int j, k;
	int len;
	cluster_node *table[REDIS_CLUSTER_SLOTS] = { NULL };

	if (cc == NULL)
	{
		return REDIS_ERR;
	}

	if (ip == NULL || port <= 0)
	{
		__redisClusterSetError(cc,
			REDIS_ERR_OTHER, "ip or port error!");
		goto error;
	}

	if (cc->connect_timeout)
	{
		c = redisConnectWithTimeout(ip, port, *cc->connect_timeout);
	}
	else
	{
		c = redisConnect(ip, port);
	}

	if (c == NULL)
	{
		__redisClusterSetError(cc, REDIS_ERR_OTHER,
			"init redis context error(return NULL)");
		goto error;
	}
	else if (c->err)
	{
		__redisClusterSetError(cc, c->err, c->errstr);
		goto error;
	}

	reply = redisCommand(c, REDIS_COMMAND_CLUSTER_NODES);

	if (reply == NULL)
	{
		__redisClusterSetError(cc, REDIS_ERR_OTHER,
			"command(cluster nodes) reply error(NULL)");
		goto error;
	}
	else if (reply->type != REDIS_REPLY_STRING)
	{
		if (reply->type == REDIS_REPLY_ERROR)
		{
			__redisClusterSetError(cc, REDIS_ERR_OTHER,
				reply->str);
		}
		else
		{
			__redisClusterSetError(cc, REDIS_ERR_OTHER,
				"command(cluster nodes) reply error(type is not string)");
		}

		goto error;
	}

	nodes = dictCreate(&clusterNodesDictType, NULL);

	slots = hiarray_create(10, sizeof(cluster_slot*));
	if (slots == NULL)
	{
		__redisClusterSetError(cc, REDIS_ERR_OTHER,
			"array create error");
		goto error;
	}

	start = reply->str;
	end = start + reply->len;

	line_start = start;

	for (pos = start; pos < end; pos++)
	{
		if (*pos == '\n')
		{
			line_end = pos - 1;
			len = line_end - line_start;

			part = sdssplitlen(line_start, len + 1, " ", 1, &count_part);

			if (part == NULL || count_part < 8)
			{
				__redisClusterSetError(cc, REDIS_ERR_OTHER,
					"split cluster nodes error");
				goto error;
			}

			//the address string is ":0", skip this node.
			if (sdslen(part[1]) == 2 && strcmp(part[1], ":0") == 0)
			{
				sdsfreesplitres(part, count_part);
				count_part = 0;
				part = NULL;

				start = pos + 1;
				line_start = start;
				pos = start;

				continue;
			}

			if (sdslen(part[2]) >= 7 && memcmp(part[2], "myself,", 7) == 0)
			{
				role_len = sdslen(part[2]) - 7;
				role = part[2] + 7;
				myself = 1;
			}
			else
			{
				role_len = sdslen(part[2]);
				role = part[2];
			}

			//add master node
			if (role_len >= 6 && memcmp(role, "master", 6) == 0)
			{
				if (count_part < 8)
				{
					__redisClusterSetError(cc, REDIS_ERR_OTHER,
						"master node part number error");
					goto error;
				}

				master = node_get_with_nodes(cc,
					part, count_part, REDIS_ROLE_MASTER);
				if (master == NULL)
				{
					goto error;
				}

				ret = dictAdd(nodes,
					sdsnewlen(master->addr, sdslen(master->addr)), master);
				if (ret != DICT_OK)
				{
					__redisClusterSetError(cc, REDIS_ERR_OTHER,
						"the address already exists in the nodes");
					cluster_node_deinit(master);
					hi_free(master);
					goto error;
				}

				if (cc->flags & HIRCLUSTER_FLAG_ADD_SLAVE)
				{
					ret = cluster_master_slave_mapping_with_name(cc,
						&nodes_name, master, master->name);
					if (ret != REDIS_OK)
					{
						cluster_node_deinit(master);
						hi_free(master);
						goto error;
					}
				}

				if (myself == 1)
				{
					master->con = c;
					c = NULL;
				}

				for (k = 8; k < count_part; k++)
				{
					slot_start_end = sdssplitlen(part[k],
						sdslen(part[k]), "-", 1, &count_slot_start_end);

					if (slot_start_end == NULL)
					{
						__redisClusterSetError(cc, REDIS_ERR_OTHER,
							"split slot start end error(NULL)");
						goto error;
					}
					else if (count_slot_start_end == 1)
					{
						slot_start =
							hi_atoi(slot_start_end[0], sdslen(slot_start_end[0]));
						slot_end = slot_start;
					}
					else if (count_slot_start_end == 2)
					{
						slot_start =
							hi_atoi(slot_start_end[0], sdslen(slot_start_end[0]));;
						slot_end =
							hi_atoi(slot_start_end[1], sdslen(slot_start_end[1]));;
					}
					else
					{
						slot_start = -1;
						slot_end = -1;
					}

					sdsfreesplitres(slot_start_end, count_slot_start_end);
					count_slot_start_end = 0;
					slot_start_end = NULL;

					if (slot_start < 0 || slot_end < 0 ||
						slot_start > slot_end || slot_end >= REDIS_CLUSTER_SLOTS)
					{
						continue;
					}

					for (j = slot_start; j <= slot_end; j++)
					{
						if (table[j] != NULL)
						{
							__redisClusterSetError(cc, REDIS_ERR_OTHER,
								"diffent node hold a same slot");
							goto error;
						}
						table[j] = master;
					}

					slot = hiarray_push(slots);
					if (slot == NULL)
					{
						__redisClusterSetError(cc, REDIS_ERR_OTHER,
							"slot push in array error");
						goto error;
					}

					*slot = cluster_slot_create(master);
					if (*slot == NULL)
					{
						__redisClusterSetError(cc, REDIS_ERR_OOM,
							"Out of memory");
						goto error;
					}

					(*slot)->start = (uint32_t)slot_start;
					(*slot)->end = (uint32_t)slot_end;
				}

			}
			//add slave node
			else if ((cc->flags & HIRCLUSTER_FLAG_ADD_SLAVE) &&
				(role_len >= 5 && memcmp(role, "slave", 5) == 0))
			{
				slave = node_get_with_nodes(cc, part,
					count_part, REDIS_ROLE_SLAVE);
				if (slave == NULL)
				{
					goto error;
				}

				ret = cluster_master_slave_mapping_with_name(cc,
					&nodes_name, slave, part[3]);
				if (ret != REDIS_OK)
				{
					cluster_node_deinit(slave);
					hi_free(slave);
					goto error;
				}

				if (myself == 1)
				{
					slave->con = c;
					c = NULL;
				}
			}

			if (myself == 1)
			{
				myself = 0;
			}

			sdsfreesplitres(part, count_part);
			count_part = 0;
			part = NULL;

			start = pos + 1;
			line_start = start;
			pos = start;
		}
	}

	if (cc->slots != NULL)
	{
		cc->slots->nelem = 0;
		hiarray_destroy(cc->slots);
		cc->slots = NULL;
	}
	cc->slots = slots;

	cluster_nodes_swap_ctx(cc->nodes, nodes);

	if (cc->nodes != NULL)
	{
		dictRelease(cc->nodes);
		cc->nodes = NULL;
	}
	cc->nodes = nodes;

	hiarray_sort(cc->slots, cluster_slot_start_cmp);

	memcpy(cc->table, table, REDIS_CLUSTER_SLOTS * sizeof(cluster_node *));
	cc->route_version++;

	freeReplyObject(reply);

	if (c != NULL)
	{
		redisFree(c);
	}

	if (nodes_name != NULL)
	{
		dictRelease(nodes_name);
	}

	return REDIS_OK;

error:

	if (part != NULL)
	{
		sdsfreesplitres(part, count_part);
		count_part = 0;
		part = NULL;
	}

	if (slot_start_end != NULL)
	{
		sdsfreesplitres(slot_start_end, count_slot_start_end);
		count_slot_start_end = 0;
		slot_start_end = NULL;
	}

	if (slots != NULL)
	{
		if (slots == cc->slots)
		{
			cc->slots = NULL;
		}

		slots->nelem = 0;
		hiarray_destroy(slots);
	}

	if (nodes != NULL)
	{
		if (nodes == cc->nodes)
		{
			cc->nodes = NULL;
		}

		dictRelease(nodes);
	}

	if (nodes_name != NULL)
	{
		dictRelease(nodes_name);
	}

	if (reply != NULL)
	{
		freeReplyObject(reply);
		reply = NULL;
	}

	if (c != NULL)
	{
		redisFree(c);
	}

	return REDIS_ERR;
}

int
cluster_update_route(redisClusterContext *cc)
{
	int ret;
	int flag_err_not_set = 1;
	cluster_node *node;
	dictIterator *it;
	dictEntry *de;

	if (cc == NULL)
	{
		return REDIS_ERR;
	}

	if (cc->ip != NULL && cc->port > 0)
	{
		ret = cluster_update_route_by_addr(cc, cc->ip, cc->port);
		if (ret == REDIS_OK)
		{
			return REDIS_OK;
		}

		flag_err_not_set = 0;
	}

	if (cc->nodes == NULL)
	{
		if (flag_err_not_set)
		{
			__redisClusterSetError(cc, REDIS_ERR_OTHER, "no server address");
		}

		return REDIS_ERR;
	}

	it = dictGetIterator(cc->nodes);
	while ((de = dictNext(it)) != NULL)
	{
		node = dictGetEntryVal(de);
		if (node == NULL || node->host == NULL || node->port < 0)
		{
			continue;
		}

		ret = cluster_update_route_by_addr(cc, node->host, node->port);
		if (ret == REDIS_OK)
		{
			if (cc->err)
			{
				cc->err = 0;
				// memset(cc->errstr, '\0', strlen(cc->errstr));
				cc->errstr[0] = '\0';
			}

			dictReleaseIterator(it);

			return REDIS_OK;
		}

		flag_err_not_set = 0;
	}

	dictReleaseIterator(it);

	if (flag_err_not_set)
	{
		__redisClusterSetError(cc, REDIS_ERR_OTHER, "no valid server address");
	}

	return REDIS_ERR;
}

static void print_cluster_node_list(redisClusterContext *cc)
{
	dictIterator *di = NULL;
	dictEntry *de;
	listIter *it;
	listNode *ln;
	cluster_node *master, *slave;
	hilist *slaves;

	if (cc == NULL)
	{
		return;
	}

	di = dictGetIterator(cc->nodes);

	printf("name\taddress\trole\tslaves\n");

	while ((de = dictNext(di)) != NULL) {
		master = dictGetEntryVal(de);

		printf("%s\t%s\t%d\t%s\n", master->name, master->addr,
			master->role, master->slaves ? "hava" : "null");

		slaves = master->slaves;
		if (slaves == NULL)
		{
			continue;
		}

		it = listGetIterator(slaves, AL_START_HEAD);
		while ((ln = listNext(it)) != NULL)
		{
			slave = listNodeValue(ln);
			printf("%s\t%s\t%d\t%s\n", slave->name, slave->addr,
				slave->role, slave->slaves ? "hava" : "null");
		}

		listReleaseIterator(it);

		printf("\n");
	}
}


int test_cluster_update_route(redisClusterContext *cc)
{
	int ret;

	ret = cluster_update_route(cc);

	//print_cluster_node_list(cc);

	return ret;
}

redisClusterContext *redisClusterContextInit(void) {
	redisClusterContext *cc;

	cc = calloc(1, sizeof(redisClusterContext));
	if (cc == NULL)
		return NULL;

	//cc->err = 0;
	//cc->errstr[0] = '\0';
	//cc->ip = NULL;
	//cc->port = 0;
	//cc->password[0] = '\0';
	//cc->flags = 0;
	//cc->connect_timeout = NULL;
	//cc->timeout = NULL;
	//cc->nodes = NULL;
	//cc->slots = NULL;
	cc->max_redirect_count = CLUSTER_DEFAULT_MAX_REDIRECT_COUNT;
	//cc->retry_count = 0;
	//cc->requests = NULL;
	//cc->need_update_route = 0;
	//cc->update_route_time = 0LL;

	//cc->route_version = 0LL;

	// memset(cc->table, 0, REDIS_CLUSTER_SLOTS * sizeof(cluster_node *));

	cc->flags |= REDIS_BLOCK;

	return cc;
}

void redisClusterFree(redisClusterContext *cc) {

	if (cc == NULL)
		return;

	if (cc->ip)
	{
		sdsfree(cc->ip);
		cc->ip = NULL;
	}

	if (cc->connect_timeout)
	{
		free(cc->connect_timeout);
	}

	if (cc->timeout)
	{
		free(cc->timeout);
	}

	memset(cc->table, 0, REDIS_CLUSTER_SLOTS * sizeof(cluster_node *));

	if (cc->slots != NULL)
	{
		cc->slots->nelem = 0;
		hiarray_destroy(cc->slots);
		cc->slots = NULL;
	}

	if (cc->nodes != NULL)
	{
		dictRelease(cc->nodes);
	}

	if (cc->requests != NULL)
	{
		listRelease(cc->requests);
	}

	free(cc);
}

/* Connect to a Redis cluster. On error the field error in the returned
 * context will be set to the return value of the error function.
 * When no set of reply functions is given, the default set will be used. */
static int _redisClusterConnect2(redisClusterContext *cc)
{

	if (cc->nodes == NULL || dictSize(cc->nodes) == 0)
	{
		__redisClusterSetError(cc, REDIS_ERR_OTHER, "servers address does not set up");
		return REDIS_ERR;
	}

	return cluster_update_route(cc);
}

/* Connect to a Redis cluster. On error the field error in the returned
 * context will be set to the return value of the error function.
 * When no set of reply functions is given, the default set will be used. */
static redisClusterContext *_redisClusterConnect(redisClusterContext *cc, const char *addrs) {

	int ret;

	ret = redisClusterSetOptionAddNodes(cc, addrs);
	if (ret != REDIS_OK)
	{
		return cc;
	}

	cluster_update_route(cc);

	return cc;
}

redisClusterContext *redisClusterConnect(const char *addrs, const char *password, int flags)
{
	redisClusterContext *cc;

	cc = redisClusterContextInit();

	if (cc == NULL)
	{
		return NULL;
	}

	redisClusterSetOptionPassword(cc, password);

	cc->flags |= REDIS_BLOCK;
	if (flags)
	{
		cc->flags |= flags;
	}

	return _redisClusterConnect(cc, addrs);
}

redisClusterContext *redisClusterConnectWithTimeout(
	const char *addrs, const char* password, const struct timeval tv, int flags)
{
	redisClusterContext *cc;

	cc = redisClusterContextInit();

	if (cc == NULL)
	{
		return NULL;
	}

	redisClusterSetOptionPassword(cc, password);

	cc->flags |= REDIS_BLOCK;
	if (flags)
	{
		cc->flags |= flags;
	}

	if (cc->connect_timeout == NULL)
	{
		cc->connect_timeout = malloc(sizeof(struct timeval));
	}

	memcpy(cc->connect_timeout, &tv, sizeof(struct timeval));

	return _redisClusterConnect(cc, addrs);
}

redisClusterContext *redisClusterConnectNonBlock(const char *addrs, const char *password, int flags) {

	redisClusterContext *cc;

	cc = redisClusterContextInit();

	if (cc == NULL)
	{
		return NULL;
	}

	redisClusterSetOptionPassword(cc, password);

	cc->flags &= ~REDIS_BLOCK;
	if (flags)
	{
		cc->flags |= flags;
	}

	return _redisClusterConnect(cc, addrs);
}

int redisClusterSetOptionPassword(redisClusterContext* cc, const char *password) {
	if (password) {
		strncpy(cc->password, password, sizeof(cc->password));
	}
	else {
		cc->password[0] = '\0';
	}
	return REDIS_OK;
}

int redisClusterSetOptionAddNode(redisClusterContext *cc, const char *addr)
{
	dictEntry *node_entry;
	cluster_node *node;
	sds *ip_port = NULL;
	int ip_port_count = 0;
	sds ip;
	int port;
	sds addr_sds = NULL;

	if (cc == NULL)
	{
		return REDIS_ERR;
	}

	if (cc->nodes == NULL)
	{
		cc->nodes = dictCreate(&clusterNodesDictType, NULL);
		if (cc->nodes == NULL)
		{
			return REDIS_ERR;
		}
	}

	addr_sds = sdsnew(addr);
	node_entry = dictFind(cc->nodes, addr_sds);
	sdsfree(addr_sds);
	if (node_entry == NULL)
	{
		ip_port = sdssplitlen(addr, strlen(addr),
			IP_PORT_SEPARATOR, strlen(IP_PORT_SEPARATOR), &ip_port_count);
		if (ip_port == NULL || ip_port_count != 2 ||
			sdslen(ip_port[0]) <= 0 || sdslen(ip_port[1]) <= 0)
		{
			if (ip_port != NULL)
			{
				sdsfreesplitres(ip_port, ip_port_count);
			}
			__redisClusterSetError(cc, REDIS_ERR_OTHER, "server address is error(correct is like: 127.0.0.1:1234)");
			return REDIS_ERR;
		}

		ip = ip_port[0];
		// port = hi_atoi(ip_port[1], sdslen(ip_port[1]));
		port = atoi(ip_port[1]);

		if (port <= 0)
		{
			sdsfreesplitres(ip_port, ip_port_count);
			__redisClusterSetError(cc, REDIS_ERR_OTHER, "server port is error");
			return REDIS_ERR;
		}

		sdsfree(ip_port[1]);
		free(ip_port);
		ip_port = NULL;

		node = hi_alloc(sizeof(cluster_node));
		if (node == NULL)
		{
			sdsfree(ip);
			__redisClusterSetError(cc, REDIS_ERR_OTHER, "alloc cluster node error");
			return REDIS_ERR;
		}

		cluster_node_init(node);

		node->addr = sdsnew(addr);
		if (node->addr == NULL)
		{
			sdsfree(ip);
			hi_free(node);
			__redisClusterSetError(cc, REDIS_ERR_OTHER, "new node address error");
			return REDIS_ERR;
		}

		node->host = ip;
		node->port = port;

		dictAdd(cc->nodes, sdsnewlen(node->addr, sdslen(node->addr)), node);
	}

	return REDIS_OK;
}

int redisClusterSetOptionAddNodes(redisClusterContext *cc, const char *addrs)
{
	int ret;
	sds *address = NULL;
	int address_count = 0;
	int i;

	if (cc == NULL)
	{
		return REDIS_ERR;
	}

	address = sdssplitlen(addrs, strlen(addrs), CLUSTER_ADDRESS_SEPARATOR,
		strlen(CLUSTER_ADDRESS_SEPARATOR), &address_count);
	if (address == NULL || address_count <= 0)
	{
		__redisClusterSetError(cc, REDIS_ERR_OTHER, "servers address is error(correct is like: 127.0.0.1:1234,127.0.0.2:5678)");
		return REDIS_ERR;
	}

	for (i = 0; i < address_count; i++)
	{
		ret = redisClusterSetOptionAddNode(cc, address[i]);
		if (ret != REDIS_OK)
		{
			sdsfreesplitres(address, address_count);
			return REDIS_ERR;
		}
	}

	sdsfreesplitres(address, address_count);

	return REDIS_OK;
}

int redisClusterSetOptionConnectBlock(redisClusterContext *cc)
{

	if (cc == NULL)
	{
		return REDIS_ERR;
	}

	cc->flags |= REDIS_BLOCK;

	return REDIS_OK;
}

int redisClusterSetOptionConnectNonBlock(redisClusterContext *cc)
{

	if (cc == NULL)
	{
		return REDIS_ERR;
	}

	cc->flags &= ~REDIS_BLOCK;

	return REDIS_OK;
}

int redisClusterSetOptionParseSlaves(redisClusterContext *cc)
{

	if (cc == NULL)
	{
		return REDIS_ERR;
	}

	cc->flags |= HIRCLUSTER_FLAG_ADD_SLAVE;

	return REDIS_OK;
}

int redisClusterSetOptionParseOpenSlots(redisClusterContext *cc)
{

	if (cc == NULL)
	{
		return REDIS_ERR;
	}

	cc->flags |= HIRCLUSTER_FLAG_ADD_OPENSLOT;

	return REDIS_OK;
}

int redisClusterSetOptionRouteUseSlots(redisClusterContext *cc)
{

	if (cc == NULL)
	{
		return REDIS_ERR;
	}

	cc->flags |= HIRCLUSTER_FLAG_ROUTE_USE_SLOTS;

	return REDIS_OK;
}

int redisClusterSetOptionConnectTimeout(redisClusterContext *cc, const struct timeval tv)
{

	if (cc == NULL)
	{
		return REDIS_ERR;
	}

	if (cc->connect_timeout == NULL)
	{
		cc->connect_timeout = malloc(sizeof(struct timeval));
	}

	memcpy(cc->connect_timeout, &tv, sizeof(struct timeval));

	return REDIS_OK;
}

int redisClusterSetOptionTimeout(redisClusterContext *cc, const struct timeval tv)
{

	if (cc == NULL)
	{
		return REDIS_ERR;
	}

	if (cc->timeout == NULL)
	{
		cc->timeout = malloc(sizeof(struct timeval));
		memcpy(cc->timeout, &tv, sizeof(struct timeval));
	}
	else if (cc->timeout->tv_sec != tv.tv_sec || cc->timeout->tv_usec != tv.tv_usec)
	{
		memcpy(cc->timeout, &tv, sizeof(struct timeval));

		if (cc->nodes && dictSize(cc->nodes) > 0)
		{
			dictEntry *de;
			dictIterator *di;
			cluster_node *node;

			di = dictGetIterator(cc->nodes);

			while ((de = dictNext(di)) != NULL)
			{
				node = dictGetEntryVal(de);
				if (node->con && node->con->flags&REDIS_CONNECTED && node->con->err == 0)
				{
					redisSetTimeout(node->con, tv);
				}

				if (node->slaves && listLength(node->slaves) > 0)
				{
					cluster_node *slave;
					listIter *li;
					listNode *ln;

					li = listGetIterator(node->slaves, AL_START_HEAD);
					while ((ln = listNext(li)) != NULL)
					{
						slave = listNodeValue(ln);
						if (slave->con && slave->con->flags&REDIS_CONNECTED && slave->con->err == 0)
						{
							redisSetTimeout(slave->con, tv);
						}
					}

					listReleaseIterator(li);
				}
			}

			dictReleaseIterator(di);
		}
	}

	return REDIS_OK;
}

int redisClusterSetOptionMaxRedirect(redisClusterContext *cc, int max_redirect_count)
{
	if (cc == NULL || max_redirect_count <= 0)
	{
		return REDIS_ERR;
	}

	cc->max_redirect_count = max_redirect_count;

	return REDIS_OK;
}

int redisClusterConnect2(redisClusterContext *cc)
{

	if (cc == NULL)
	{
		return REDIS_ERR;
	}

	return _redisClusterConnect2(cc);
}

redisContext *ctx_get_by_node(redisClusterContext *cc, cluster_node *node)
{
	redisContext *c = NULL;
	if (node == NULL)
	{
		return NULL;
	}

	c = node->con;
	if (c != NULL)
	{
		if (c->err)
		{
			redisReconnect(c);
			if (cc->timeout && c->err == 0) {
				redisSetTimeout(c, *cc->timeout);
			}

			if (_redisClusterAuth(cc, c) != REDIS_OK) {
				redisFree(c);
				return NULL;
			}
		}
		return c;
	}

	if (node->host == NULL || node->port <= 0)
	{
		return NULL;
	}

	if (cc->connect_timeout)
	{
		c = redisConnectWithTimeout(node->host, node->port, *cc->connect_timeout);
	}
	else
	{
		c = redisConnect(node->host, node->port);
	}

	if (cc->timeout && c != NULL && c->err == 0) {
		redisSetTimeout(c, *cc->timeout);
	}

	if (_redisClusterAuth(cc, c) != REDIS_OK) {
		redisFree(c);
		return NULL;
	}

	node->con = c;
	return c;
}

static cluster_node *node_get_by_slot(redisClusterContext *cc, uint32_t slot_num)
{
	struct hiarray *slots;
	uint32_t slot_count;
	cluster_slot **slot;
	uint32_t middle, start, end;
	uint8_t stop = 0;

	if (cc == NULL)
	{
		return NULL;
	}

	if (slot_num >= REDIS_CLUSTER_SLOTS)
	{
		return NULL;
	}

	slots = cc->slots;
	if (slots == NULL)
	{
		return NULL;
	}
	slot_count = hiarray_n(slots);

	start = 0;
	end = slot_count - 1;
	middle = 0;

	do {
		if (start >= end)
		{
			stop = 1;
			middle = end;
		}
		else
		{
			middle = start + (end - start) / 2;
		}

		ASSERT(middle < slot_count);

		slot = hiarray_get(slots, middle);
		if ((*slot)->start > slot_num)
		{
			end = middle - 1;
		}
		else if ((*slot)->end < slot_num)
		{
			start = middle + 1;
		}
		else
		{
			return (*slot)->node;
		}


	} while (!stop);

	printf("slot_num : %d\n", slot_num);
	printf("slot_count : %d\n", slot_count);
	printf("start : %d\n", start);
	printf("end : %d\n", end);
	printf("middle : %d\n", middle);

	return NULL;
}


static cluster_node *node_get_by_table(redisClusterContext *cc, uint32_t slot_num)
{
	if (cc == NULL)
	{
		return NULL;
	}

	if (slot_num >= REDIS_CLUSTER_SLOTS)
	{
		return NULL;
	}

	return cc->table[slot_num];

}

static cluster_node *node_get_witch_connected(redisClusterContext *cc)
{
	dictIterator *di;
	dictEntry *de;
	struct cluster_node *node;
	redisContext *c = NULL;
	redisReply *reply = NULL;

	if (cc == NULL || cc->nodes == NULL)
	{
		return NULL;
	}

	di = dictGetIterator(cc->nodes);
	while ((de = dictNext(di)) != NULL)
	{
		node = dictGetEntryVal(de);
		if (node == NULL)
		{
			continue;
		}

		c = ctx_get_by_node(cc, node);
		if (c == NULL || c->err)
		{
			continue;
		}

		reply = redisCommand(c, REDIS_COMMAND_PING);
		if (reply != NULL && reply->type == REDIS_REPLY_STATUS &&
			reply->str != NULL && strcmp(reply->str, "PONG") == 0)
		{
			freeReplyObject(reply);
			reply = NULL;

			dictReleaseIterator(di);

			return node;
		}
		else if (reply != NULL)
		{
			freeReplyObject(reply);
			reply = NULL;
		}
	}

	dictReleaseIterator(di);

	return NULL;
}

static int slot_get_by_command(redisClusterContext *cc, char *cmd, int len)
{
	struct cmd *command = NULL;
	struct keypos *kp;
	int key_count;
	uint32_t i;
	int slot_num = -1;

	if (cc == NULL || cmd == NULL || len <= 0)
	{
		goto done;
	}

	command = command_get();
	if (command == NULL)
	{
		__redisClusterSetError(cc, REDIS_ERR_OOM, "Out of memory");
		goto done;
	}

	command->cmd = cmd;
	command->clen = len;
	redis_parse_cmd(command);
	if (command->result != CMD_PARSE_OK)
	{
		__redisClusterSetError(cc, REDIS_ERR_PROTOCOL, "parse command error");
		goto done;
	}

	key_count = hiarray_n(command->keys);

	if (key_count <= 0)
	{
		__redisClusterSetError(cc, REDIS_ERR_OTHER, "no keys in command(must have keys for redis cluster mode)");
		goto done;
	}
	else if (key_count == 1)
	{
		kp = hiarray_get(command->keys, 0);
		slot_num = keyHashSlot(kp->start, kp->end - kp->start);

		goto done;
	}

	for (i = 0; i < hiarray_n(command->keys); i++)
	{
		kp = hiarray_get(command->keys, i);

		slot_num = keyHashSlot(kp->start, kp->end - kp->start);
	}

done:

	if (command != NULL)
	{
		command->cmd = NULL;
		command_destroy(command);
	}

	return slot_num;
}

/* Get the cluster config from one node.
  * Return value: config_value string must free by usr.
  */
static char * cluster_config_get(redisClusterContext *cc,
	const char *config_name, int *config_value_len)
{
	redisContext *c;
	cluster_node *node;
	redisReply *reply = NULL, *sub_reply;
	char *config_value = NULL;

	if (cc == NULL || config_name == NULL
		|| config_value_len == NULL)
	{
		return NULL;
	}

	node = node_get_witch_connected(cc);
	if (node == NULL)
	{
		__redisClusterSetError(cc,
			REDIS_ERR_OTHER, "no reachable node in cluster");
		goto error;
	}

	c = ctx_get_by_node(cc, node);

	reply = redisCommand(c, "config get %s", config_name);
	if (reply == NULL)
	{
		__redisClusterSetError(cc,
			REDIS_ERR_OTHER, "reply for config get is null");
		goto error;
	}

	if (reply->type != REDIS_REPLY_ARRAY)
	{
		__redisClusterSetError(cc, REDIS_ERR_OTHER,
			"reply for config get type is not array");
		goto error;
	}

	if (reply->elements != 2)
	{
		__redisClusterSetError(cc, REDIS_ERR_OTHER,
			"reply for config get elements number is not 2");
		goto error;
	}

	sub_reply = reply->element[0];
	if (sub_reply == NULL || sub_reply->type != REDIS_REPLY_STRING)
	{
		__redisClusterSetError(cc, REDIS_ERR_OTHER,
			"reply for config get config name is not string");
		goto error;
	}

	if (strcmp(sub_reply->str, config_name))
	{
		__redisClusterSetError(cc, REDIS_ERR_OTHER,
			"reply for config get config name is not we want");
		goto error;
	}

	sub_reply = reply->element[1];
	if (sub_reply == NULL || sub_reply->type != REDIS_REPLY_STRING)
	{
		__redisClusterSetError(cc, REDIS_ERR_OTHER,
			"reply for config get config value type is not string");
		goto error;
	}

	config_value = sub_reply->str;
	*config_value_len = sub_reply->len;
	sub_reply->str = NULL;

	if (reply != NULL)
	{
		freeReplyObject(reply);
	}

	return config_value;

error:

	if (reply != NULL)
	{
		freeReplyObject(reply);
	}

	return NULL;
}

/* Helper function for the redisClusterAppendCommand* family of functions.
 *
 * Write a formatted command to the output buffer. When this family
 * is used, you need to call redisGetReply yourself to retrieve
 * the reply (or replies in pub/sub).
 */
static int __redisClusterAppendCommand(redisClusterContext *cc,
	struct cmd *command) {

	cluster_node *node;
	redisContext *c = NULL;

	if (cc == NULL || command == NULL)
	{
		return REDIS_ERR;
	}

	node = node_get_by_table(cc, (uint32_t)command->slot_num);
	if (node == NULL)
	{
		__redisClusterSetError(cc, REDIS_ERR_OTHER, "node get by slot error");
		return REDIS_ERR;
	}

	c = ctx_get_by_node(cc, node);
	if (c == NULL)
	{
		__redisClusterSetError(cc, REDIS_ERR_OTHER, "ctx get by node is null");
		return REDIS_ERR;
	}
	else if (c->err)
	{
		__redisClusterSetError(cc, c->err, c->errstr);
		return REDIS_ERR;
	}

	if (__redisAppendCommand(c, command->cmd, command->clen) != REDIS_OK)
	{
		__redisClusterSetError(cc, c->err, c->errstr);
		return REDIS_ERR;
	}

	return REDIS_OK;
}

/* Helper function for the redisClusterGetReply* family of functions.
 */
static int __redisClusterGetReply(redisClusterContext *cc, int slot_num, void **reply)
{
	cluster_node *node;
	redisContext *c;

	if (cc == NULL || slot_num < 0 || reply == NULL)
	{
		return REDIS_ERR;
	}

	node = node_get_by_table(cc, (uint32_t)slot_num);
	if (node == NULL)
	{
		__redisClusterSetError(cc, REDIS_ERR_OTHER, "node get by table is null");
		return REDIS_ERR;
	}

	c = ctx_get_by_node(cc, node);
	if (c == NULL)
	{
		__redisClusterSetError(cc, REDIS_ERR_OOM, "Out of memory");
		return REDIS_ERR;
	}
	else if (c->err)
	{
		if (cc->need_update_route == 0)
		{
			cc->retry_count++;
			if (cc->retry_count > cc->max_redirect_count)
			{
				cc->need_update_route = 1;
				cc->retry_count = 0;
			}
		}
		__redisClusterSetError(cc, c->err, c->errstr);
		return REDIS_ERR;
	}

	if (redisGetReply(c, reply) != REDIS_OK)
	{
		__redisClusterSetError(cc, c->err, c->errstr);
		return REDIS_ERR;
	}

	if (cluster_reply_error_type(*reply) == CLUSTER_ERR_MOVED)
	{
		cc->need_update_route = 1;
	}

	return REDIS_OK;
}

static cluster_node *node_get_by_ask_error_reply(
	redisClusterContext *cc, redisReply *reply)
{
	sds *part = NULL, *ip_port = NULL;
	int part_len = 0, ip_port_len;
	dictEntry *de;
	cluster_node *node = NULL;

	if (cc == NULL || reply == NULL)
	{
		return NULL;
	}

	if (cluster_reply_error_type(reply) != CLUSTER_ERR_ASK)
	{
		__redisClusterSetError(cc, REDIS_ERR_OTHER,
			"reply is not ask error!");
		return NULL;
	}

	part = sdssplitlen(reply->str, reply->len, " ", 1, &part_len);

	if (part != NULL && part_len == 3)
	{
		ip_port = sdssplitlen(part[2], sdslen(part[2]),
			":", 1, &ip_port_len);

		if (ip_port != NULL && ip_port_len == 2)
		{
			de = dictFind(cc->nodes, part[2]);
			if (de == NULL)
			{
				node = hi_alloc(sizeof(cluster_node));
				if (node == NULL)
				{
					__redisClusterSetError(cc,
						REDIS_ERR_OOM, "Out of memory");

					goto done;
				}

				cluster_node_init(node);
				node->addr = part[1];
				node->host = ip_port[0];
				// node->port = hi_atoi(ip_port[1], sdslen(ip_port[1]));
				node->port = atoi(ip_port[1]);
				node->role = REDIS_ROLE_MASTER;

				dictAdd(cc->nodes, sdsnewlen(node->addr, sdslen(node->addr)), node);

				part = NULL;
				ip_port = NULL;
			}
			else
			{
				node = de->val;

				goto done;
			}
		}
		else
		{
			__redisClusterSetError(cc, REDIS_ERR_OTHER,
				"ask error reply address part parse error!");

			goto done;
		}

	}
	else
	{
		__redisClusterSetError(cc, REDIS_ERR_OTHER,
			"ask error reply parse error!");

		goto done;
	}

done:

	if (part != NULL)
	{
		sdsfreesplitres(part, part_len);
		part = NULL;
	}

	if (ip_port != NULL)
	{
		sdsfreesplitres(ip_port, ip_port_len);
		ip_port = NULL;
	}

	return node;
}

static void *redis_cluster_command_execute(redisClusterContext *cc,
	struct cmd *command)
{
	int ret;
	void *reply = NULL;
	cluster_node *node;
	redisContext *c = NULL;
	int error_type;

retry:

	node = node_get_by_table(cc, (uint32_t)command->slot_num);
	if (node == NULL)
	{
		__redisClusterSetError(cc, REDIS_ERR_OTHER, "node get by table error");
		return NULL;
	}

	c = ctx_get_by_node(cc, node);
	if (c == NULL)
	{
		__redisClusterSetError(cc, REDIS_ERR_OTHER, "ctx get by node is null");
		return NULL;
	}
	else if (c->err)
	{
		node = node_get_witch_connected(cc);
		if (node == NULL)
		{
			__redisClusterSetError(cc, REDIS_ERR_OTHER, "no reachable node in cluster");
			return NULL;
		}

		cc->retry_count++;
		if (cc->retry_count > cc->max_redirect_count)
		{
			__redisClusterSetError(cc, REDIS_ERR_CLUSTER_TOO_MANY_REDIRECT,
				"too many cluster redirect");
			return NULL;
		}

		c = ctx_get_by_node(cc, node);
		if (c == NULL)
		{
			__redisClusterSetError(cc, REDIS_ERR_OTHER, "ctx get by node error");
			return NULL;
		}
		else if (c->err)
		{
			__redisClusterSetError(cc, c->err, c->errstr);
			return NULL;
		}
	}

ask_retry:

	if (__redisAppendCommand(c, command->cmd, command->clen) != REDIS_OK)
	{
		__redisClusterSetError(cc, c->err, c->errstr);
		return NULL;
	}

	reply = __redisBlockForReply(c);
	if (reply == NULL)
	{
		__redisClusterSetError(cc, c->err, c->errstr);
		return NULL;
	}

	error_type = cluster_reply_error_type(reply);
	if (error_type > CLUSTER_NOT_ERR && error_type < CLUSTER_ERR_SENTINEL)
	{
		cc->retry_count++;
		if (cc->retry_count > cc->max_redirect_count)
		{
			__redisClusterSetError(cc, REDIS_ERR_CLUSTER_TOO_MANY_REDIRECT,
				"too many cluster redirect");
			freeReplyObject(reply);
			return NULL;
		}

		switch (error_type)
		{
		case CLUSTER_ERR_MOVED:
			freeReplyObject(reply);
			reply = NULL;
			ret = cluster_update_route(cc);
			if (ret != REDIS_OK)
			{
				__redisClusterSetError(cc, REDIS_ERR_OTHER,
					"route update error, please recreate redisClusterContext!");
				return NULL;
			}

			goto retry;

			break;
		case CLUSTER_ERR_ASK:
			node = node_get_by_ask_error_reply(cc, reply);
			if (node == NULL)
			{
				freeReplyObject(reply);
				return NULL;
			}

			freeReplyObject(reply);
			reply = NULL;

			c = ctx_get_by_node(cc, node);
			if (c == NULL)
			{
				__redisClusterSetError(cc, REDIS_ERR_OTHER, "ctx get by node error");
				return NULL;
			}
			else if (c->err)
			{
				__redisClusterSetError(cc, c->err, c->errstr);
				return NULL;
			}

			reply = redisCommand(c, REDIS_COMMAND_ASKING);
			if (reply == NULL)
			{
				__redisClusterSetError(cc, c->err, c->errstr);
				return NULL;
			}

			freeReplyObject(reply);
			reply = NULL;

			goto ask_retry;

			break;
		case CLUSTER_ERR_TRYAGAIN:
		case CLUSTER_ERR_CROSSSLOT:
		case CLUSTER_ERR_CLUSTERDOWN:
			freeReplyObject(reply);
			reply = NULL;
			goto retry;

			break;
		default:

			break;
		}
	}

	return reply;
}

static int command_pre_fragment(redisClusterContext *cc,
	struct cmd *command, hilist *commands)
{

	struct keypos *kp, *sub_kp;
	uint32_t key_count;
	uint32_t i, j;
	uint32_t idx;
	uint32_t key_len;
	int slot_num = -1;
	struct cmd *sub_command;
	struct cmd **sub_commands = NULL;
	char num_str[12];
	uint8_t num_str_len;


	if (command == NULL || commands == NULL)
	{
		goto done;
	}

	key_count = hiarray_n(command->keys);

	sub_commands = hi_zalloc(REDIS_CLUSTER_SLOTS * sizeof(*sub_commands));
	if (sub_commands == NULL)
	{
		__redisClusterSetError(cc, REDIS_ERR_OOM, "Out of memory");
		goto done;
	}

	command->frag_seq = hi_alloc(key_count * sizeof(*command->frag_seq));
	if (command->frag_seq == NULL)
	{
		__redisClusterSetError(cc, REDIS_ERR_OOM, "Out of memory");
		goto done;
	}


	for (i = 0; i < key_count; i++)
	{
		kp = hiarray_get(command->keys, i);

		slot_num = keyHashSlot(kp->start, kp->end - kp->start);

		if (slot_num < 0 || slot_num >= REDIS_CLUSTER_SLOTS)
		{
			__redisClusterSetError(cc, REDIS_ERR_OTHER, "keyHashSlot return error");
			goto done;
		}

		if (sub_commands[slot_num] == NULL) {
			sub_commands[slot_num] = command_get();
			if (sub_commands[slot_num] == NULL) {
				__redisClusterSetError(cc, REDIS_ERR_OOM, "Out of memory");
				slot_num = -1;
				goto done;
			}
		}

		command->frag_seq[i] = sub_command = sub_commands[slot_num];

		sub_command->narg++;

		sub_kp = hiarray_push(sub_command->keys);
		if (sub_kp == NULL) {
			__redisClusterSetError(cc, REDIS_ERR_OOM, "Out of memory");
			slot_num = -1;
			goto done;
		}

		sub_kp->start = kp->start;
		sub_kp->end = kp->end;

		key_len = (uint32_t)(kp->end - kp->start);

		sub_command->clen += key_len + uint_len(key_len);

		sub_command->slot_num = slot_num;

		if (command->type == CMD_REQ_REDIS_MSET) {
			uint32_t len = 0;
			char *p;

			for (p = sub_kp->end + 1; !isdigit(*p); p++) {}

			p = sub_kp->end + 1;
			while (!isdigit(*p))
			{
				p++;
			}

			for (; isdigit(*p); p++) {
				len = len * 10 + (uint32_t)(*p - '0');
			}

			len += CRLF_LEN * 2;
			len += (p - sub_kp->end);
			sub_kp->remain_len = len;
			sub_command->clen += len;
		}
	}

	for (i = 0; i < REDIS_CLUSTER_SLOTS; i++) {     /* prepend command header */
		sub_command = sub_commands[i];
		if (sub_command == NULL) {
			continue;
		}

		idx = 0;
		if (command->type == CMD_REQ_REDIS_MGET) {
			//"*%d\r\n$4\r\nmget\r\n"

			sub_command->clen += 5 * sub_command->narg;

			sub_command->narg++;

			hi_itoa(num_str, sub_command->narg);
			num_str_len = (uint8_t)(strlen(num_str));

			sub_command->clen += 13 + num_str_len;

			sub_command->cmd = hi_zalloc(sub_command->clen * sizeof(*sub_command->cmd));
			if (sub_command->cmd == NULL)
			{
				__redisClusterSetError(cc, REDIS_ERR_OOM, "Out of memory");
				slot_num = -1;
				goto done;
			}

			sub_command->cmd[idx++] = '*';
			memcpy(sub_command->cmd + idx, num_str, num_str_len);
			idx += num_str_len;
			memcpy(sub_command->cmd + idx, "\r\n$4\r\nmget\r\n", 12);
			idx += 12;

			for (j = 0; j < hiarray_n(sub_command->keys); j++)
			{
				kp = hiarray_get(sub_command->keys, j);
				key_len = (uint32_t)(kp->end - kp->start);
				hi_itoa(num_str, key_len);
				num_str_len = strlen(num_str);

				sub_command->cmd[idx++] = '$';
				memcpy(sub_command->cmd + idx, num_str, num_str_len);
				idx += num_str_len;
				memcpy(sub_command->cmd + idx, CRLF, CRLF_LEN);
				idx += CRLF_LEN;
				memcpy(sub_command->cmd + idx, kp->start, key_len);
				idx += key_len;
				memcpy(sub_command->cmd + idx, CRLF, CRLF_LEN);
				idx += CRLF_LEN;
			}
		}
		else if (command->type == CMD_REQ_REDIS_DEL) {
			//"*%d\r\n$3\r\ndel\r\n"

			sub_command->clen += 5 * sub_command->narg;

			sub_command->narg++;

			hi_itoa(num_str, sub_command->narg);
			num_str_len = (uint8_t)strlen(num_str);

			sub_command->clen += 12 + num_str_len;

			sub_command->cmd = hi_zalloc(sub_command->clen * sizeof(*sub_command->cmd));
			if (sub_command->cmd == NULL)
			{
				__redisClusterSetError(cc, REDIS_ERR_OOM, "Out of memory");
				slot_num = -1;
				goto done;
			}

			sub_command->cmd[idx++] = '*';
			memcpy(sub_command->cmd + idx, num_str, num_str_len);
			idx += num_str_len;
			memcpy(sub_command->cmd + idx, "\r\n$3\r\ndel\r\n", 11);
			idx += 11;

			for (j = 0; j < hiarray_n(sub_command->keys); j++)
			{
				kp = hiarray_get(sub_command->keys, j);
				key_len = (uint32_t)(kp->end - kp->start);
				hi_itoa(num_str, key_len);
				num_str_len = strlen(num_str);

				sub_command->cmd[idx++] = '$';
				memcpy(sub_command->cmd + idx, num_str, num_str_len);
				idx += num_str_len;
				memcpy(sub_command->cmd + idx, CRLF, CRLF_LEN);
				idx += CRLF_LEN;
				memcpy(sub_command->cmd + idx, kp->start, key_len);
				idx += key_len;
				memcpy(sub_command->cmd + idx, CRLF, CRLF_LEN);
				idx += CRLF_LEN;
			}
		}
		else if (command->type == CMD_REQ_REDIS_MSET) {
			//"*%d\r\n$4\r\nmset\r\n"

			sub_command->clen += 3 * sub_command->narg;

			sub_command->narg *= 2;

			sub_command->narg++;

			hi_itoa(num_str, sub_command->narg);
			num_str_len = (uint8_t)strlen(num_str);

			sub_command->clen += 13 + num_str_len;

			sub_command->cmd = hi_zalloc(sub_command->clen * sizeof(*sub_command->cmd));
			if (sub_command->cmd == NULL)
			{
				__redisClusterSetError(cc, REDIS_ERR_OOM, "Out of memory");
				slot_num = -1;
				goto done;
			}

			sub_command->cmd[idx++] = '*';
			memcpy(sub_command->cmd + idx, num_str, num_str_len);
			idx += num_str_len;
			memcpy(sub_command->cmd + idx, "\r\n$4\r\nmset\r\n", 12);
			idx += 12;

			for (j = 0; j < hiarray_n(sub_command->keys); j++)
			{
				kp = hiarray_get(sub_command->keys, j);
				key_len = (uint32_t)(kp->end - kp->start);
				hi_itoa(num_str, key_len);
				num_str_len = strlen(num_str);

				sub_command->cmd[idx++] = '$';
				memcpy(sub_command->cmd + idx, num_str, num_str_len);
				idx += num_str_len;
				memcpy(sub_command->cmd + idx, CRLF, CRLF_LEN);
				idx += CRLF_LEN;
				memcpy(sub_command->cmd + idx, kp->start, key_len + kp->remain_len);
				idx += key_len + kp->remain_len;

			}
		}
		else {
			NOT_REACHED();
		}

		//printf("len : %d\n", sub_command->clen);
		//print_string_with_length_fix_CRLF(sub_command->cmd, sub_command->clen);

		sub_command->type = command->type;

		listAddNodeTail(commands, sub_command);
	}

done:

	if (sub_commands != NULL)
	{
		hi_free(sub_commands);
	}

	if (slot_num >= 0 && commands != NULL
		&& listLength(commands) == 1)
	{
		listNode *list_node = listFirst(commands);
		listDelNode(commands, list_node);
		if (command->frag_seq)
		{
			hi_free(command->frag_seq);
			command->frag_seq = NULL;
		}

		command->slot_num = slot_num;
	}

	return slot_num;
}

static void *command_post_fragment(redisClusterContext *cc,
	struct cmd *command, hilist *commands)
{
	struct cmd *sub_command;
	listNode *list_node;
	listIter *list_iter;
	redisReply *reply, *sub_reply;
	long long count = 0;

	list_iter = listGetIterator(commands, AL_START_HEAD);
	while ((list_node = listNext(list_iter)) != NULL)
	{
		sub_command = list_node->value;
		reply = sub_command->reply;
		if (reply == NULL)
		{
			return NULL;
		}
		else if (reply->type == REDIS_REPLY_ERROR)
		{
			return reply;
		}

		if (command->type == CMD_REQ_REDIS_MGET) {
			if (reply->type != REDIS_REPLY_ARRAY)
			{
				__redisClusterSetError(cc, REDIS_ERR_OTHER, "reply type is error(here only can be array)");
				return NULL;
			}
		}
		else if (command->type == CMD_REQ_REDIS_DEL) {
			if (reply->type != REDIS_REPLY_INTEGER)
			{
				__redisClusterSetError(cc, REDIS_ERR_OTHER, "reply type is error(here only can be integer)");
				return NULL;
			}

			count += reply->integer;
		}
		else if (command->type == CMD_REQ_REDIS_MSET) {
			if (reply->type != REDIS_REPLY_STATUS ||
				reply->len != 2 || strcmp(reply->str, REDIS_STATUS_OK) != 0)
			{
				__redisClusterSetError(cc, REDIS_ERR_OTHER, "reply type is error(here only can be status and ok)");
				return NULL;
			}
		}
		else {
			NOT_REACHED();
		}
	}

	reply = hi_calloc(1, sizeof(*reply));

	if (reply == NULL)
	{
		__redisClusterSetError(cc, REDIS_ERR_OOM, "Out of memory");
		return NULL;
	}

	if (command->type == CMD_REQ_REDIS_MGET) {
		int i;
		uint32_t key_count;

		reply->type = REDIS_REPLY_ARRAY;

		key_count = hiarray_n(command->keys);

		reply->elements = key_count;
		reply->element = hi_calloc(key_count, sizeof(*reply));
		if (reply->element == NULL) {
			freeReplyObject(reply);
			__redisClusterSetError(cc, REDIS_ERR_OOM, "Out of memory");
			return NULL;
		}

		for (i = key_count - 1; i >= 0; i--) {      /* for each key */
			sub_reply = command->frag_seq[i]->reply;            /* get it's reply */
			if (sub_reply == NULL) {
				freeReplyObject(reply);
				__redisClusterSetError(cc, REDIS_ERR_OTHER, "sub reply is null");
				return NULL;
			}

			if (sub_reply->type == REDIS_REPLY_STRING)
			{
				reply->element[i] = sub_reply;
			}
			else if (sub_reply->type == REDIS_REPLY_ARRAY)
			{
				if (sub_reply->elements == 0)
				{
					freeReplyObject(reply);
					__redisClusterSetError(cc, REDIS_ERR_OTHER, "sub reply elements error");
					return NULL;
				}

				reply->element[i] = sub_reply->element[sub_reply->elements - 1];
				sub_reply->elements--;
			}
		}
	}
	else if (command->type == CMD_REQ_REDIS_DEL) {
		reply->type = REDIS_REPLY_INTEGER;
		reply->integer = count;
	}
	else if (command->type == CMD_REQ_REDIS_MSET) {
		reply->type = REDIS_REPLY_STATUS;
		uint32_t str_len = strlen(REDIS_STATUS_OK);
		reply->str = hi_alloc((str_len + 1) * sizeof(char*));
		if (reply->str == NULL)
		{
			freeReplyObject(reply);
			__redisClusterSetError(cc, REDIS_ERR_OOM, "Out of memory");
			return NULL;
		}

		reply->len = str_len;
		memcpy(reply->str, REDIS_STATUS_OK, str_len);
		reply->str[str_len] = '\0';
	}
	else {
		NOT_REACHED();
	}

	return reply;
}

/*
 * Split the command into subcommands by slot
 *
 * Returns slot_num
 * If slot_num < 0 or slot_num >=  REDIS_CLUSTER_SLOTS means this function runs error;
 * Otherwise if  the commands > 1 , slot_num is the last subcommand slot number.
 */
static int command_format_by_slot(redisClusterContext *cc,
	struct cmd *command, hilist *commands)
{
	struct keypos *kp;
	int key_count;
	int slot_num = -1;

	if (cc == NULL || commands == NULL ||
		command == NULL ||
		command->cmd == NULL || command->clen <= 0)
	{
		goto done;
	}


	redis_parse_cmd(command);
	if (command->result == CMD_PARSE_ENOMEM)
	{
		__redisClusterSetError(cc, REDIS_ERR_PROTOCOL, "Parse command error: out of memory");
		goto done;
	}
	else if (command->result != CMD_PARSE_OK)
	{
		__redisClusterSetError(cc, REDIS_ERR_PROTOCOL, command->errstr);
		goto done;
	}

	key_count = hiarray_n(command->keys);

	if (key_count <= 0)
	{
		__redisClusterSetError(cc, REDIS_ERR_OTHER, "No keys in command(must have keys for redis cluster mode)");
		goto done;
	}
	else if (key_count == 1)
	{
		kp = hiarray_get(command->keys, 0);
		slot_num = keyHashSlot(kp->start, kp->end - kp->start);
		command->slot_num = slot_num;

		goto done;
	}

	slot_num = command_pre_fragment(cc, command, commands);

done:

	return slot_num;
}


void redisClusterSetMaxRedirect(redisClusterContext *cc, int max_redirect_count)
{
	if (cc == NULL || max_redirect_count <= 0)
	{
		return;
	}

	cc->max_redirect_count = max_redirect_count;
}

void *redisClusterFormattedCommand(redisClusterContext *cc, char *cmd, int len) {
	redisReply *reply = NULL;
	int slot_num;
	struct cmd *command = NULL, *sub_command;
	hilist *commands = NULL;
	listNode *list_node;
	listIter *list_iter = NULL;

	if (cc == NULL)
	{
		return NULL;
	}

	if (cc->err)
	{
		cc->err = 0;
		// memset(cc->errstr, '\0', strlen(cc->errstr));
		cc->errstr[0] = '\0';
	}

	command = command_get();
	if (command == NULL)
	{
		__redisClusterSetError(cc, REDIS_ERR_OOM, "Out of memory");
		return NULL;
	}

	command->cmd = cmd;
	command->clen = len;

	commands = listCreate();
	if (commands == NULL)
	{
		__redisClusterSetError(cc, REDIS_ERR_OOM, "Out of memory");
		goto error;
	}

	commands->free = listCommandFree;

	slot_num = command_format_by_slot(cc, command, commands);

	if (slot_num < 0)
	{
		goto error;
	}
	else if (slot_num >= REDIS_CLUSTER_SLOTS)
	{
		__redisClusterSetError(cc, REDIS_ERR_OTHER, "slot_num is out of range");
		goto error;
	}

	//all keys belong to one slot
	if (listLength(commands) == 0)
	{
		reply = redis_cluster_command_execute(cc, command);
		goto done;
	}

	ASSERT(listLength(commands) != 1);

	list_iter = listGetIterator(commands, AL_START_HEAD);
	while ((list_node = listNext(list_iter)) != NULL)
	{
		sub_command = list_node->value;

		reply = redis_cluster_command_execute(cc, sub_command);
		if (reply == NULL)
		{
			goto error;
		}
		else if (reply->type == REDIS_REPLY_ERROR)
		{
			goto done;
		}

		sub_command->reply = reply;
	}

	reply = command_post_fragment(cc, command, commands);

done:

	command->cmd = NULL;
	command_destroy(command);

	if (commands != NULL)
	{
		listRelease(commands);
	}

	if (list_iter != NULL)
	{
		listReleaseIterator(list_iter);
	}

	cc->retry_count = 0;

	return reply;

error:

	if (command != NULL)
	{
		command->cmd = NULL;
		command_destroy(command);
	}

	if (commands != NULL)
	{
		listRelease(commands);
	}

	if (list_iter != NULL)
	{
		listReleaseIterator(list_iter);
	}

	cc->retry_count = 0;

	return NULL;
}

void *redisClustervCommand(redisClusterContext *cc, const char *format, va_list ap) {
	redisReply *reply;
	char *cmd;
	int len;

	if (cc == NULL)
	{
		return NULL;
	}

	len = redisvFormatCommand(&cmd, format, ap);

	if (len == -1) {
		__redisClusterSetError(cc, REDIS_ERR_OOM, "Out of memory");
		return NULL;
	}
	else if (len == -2) {
		__redisClusterSetError(cc, REDIS_ERR_OTHER, "Invalid format string");
		return NULL;
	}

	reply = redisClusterFormattedCommand(cc, cmd, len);

	free(cmd);

	return reply;
}

void *redisClusterCommand(redisClusterContext *cc, const char *format, ...) {
	va_list ap;
	redisReply *reply = NULL;

	va_start(ap, format);
	reply = redisClustervCommand(cc, format, ap);
	va_end(ap);

	return reply;
}

void *redisClusterCommandArgv(redisClusterContext *cc, int argc, const char **argv, const size_t *argvlen) {
	redisReply *reply = NULL;
	char *cmd;
	int len;

	len = redisFormatCommandArgv(&cmd, argc, argv, argvlen);
	if (len == -1) {
		__redisClusterSetError(cc, REDIS_ERR_OOM, "Out of memory");
		return NULL;
	}

	reply = redisClusterFormattedCommand(cc, cmd, len);

	free(cmd);

	return reply;
}

int redisClusterAppendFormattedCommand(redisClusterContext *cc,
	char *cmd, int len) {
	int slot_num;
	struct cmd *command = NULL, *sub_command;
	hilist *commands = NULL;
	listNode *list_node;
	listIter *list_iter = NULL;

	if (cc->requests == NULL)
	{
		cc->requests = listCreate();
		if (cc->requests == NULL)
		{
			__redisClusterSetError(cc, REDIS_ERR_OOM, "Out of memory");
			goto error;
		}

		cc->requests->free = listCommandFree;
	}

	command = command_get();
	if (command == NULL)
	{
		__redisClusterSetError(cc, REDIS_ERR_OOM, "Out of memory");
		goto error;
	}

	command->cmd = cmd;
	command->clen = len;

	commands = listCreate();
	if (commands == NULL)
	{
		__redisClusterSetError(cc, REDIS_ERR_OOM, "Out of memory");
		goto error;
	}

	commands->free = listCommandFree;

	slot_num = command_format_by_slot(cc, command, commands);

	if (slot_num < 0)
	{
		goto error;
	}
	else if (slot_num >= REDIS_CLUSTER_SLOTS)
	{
		__redisClusterSetError(cc, REDIS_ERR_OTHER, "slot_num is out of range");
		goto error;
	}

	//all keys belong to one slot
	if (listLength(commands) == 0)
	{
		if (__redisClusterAppendCommand(cc, command) == REDIS_OK)
		{
			goto done;
		}
		else
		{
			goto error;
		}
	}

	ASSERT(listLength(commands) != 1);

	list_iter = listGetIterator(commands, AL_START_HEAD);
	while ((list_node = listNext(list_iter)) != NULL)
	{
		sub_command = list_node->value;

		if (__redisClusterAppendCommand(cc, sub_command) == REDIS_OK)
		{
			continue;
		}
		else
		{
			goto error;
		}
	}

done:

	if (command->cmd != NULL)
	{
		command->cmd = NULL;
	}
	else
	{
		goto error;
	}

	if (commands != NULL)
	{
		if (listLength(commands) > 0)
		{
			command->sub_commands = commands;
		}
		else
		{
			listRelease(commands);
		}
	}

	if (list_iter != NULL)
	{
		listReleaseIterator(list_iter);
	}

	listAddNodeTail(cc->requests, command);

	return REDIS_OK;

error:

	if (command != NULL)
	{
		command->cmd = NULL;
		command_destroy(command);
	}

	if (commands != NULL)
	{
		listRelease(commands);
	}

	if (list_iter != NULL)
	{
		listReleaseIterator(list_iter);
	}

	/* Attention: mybe here we must pop the
	  sub_commands that had append to the nodes.
	  But now we do not handle it. */

	return REDIS_ERR;
}


int redisClustervAppendCommand(redisClusterContext *cc,
	const char *format, va_list ap) {
	int ret;
	char *cmd;
	int len;

	len = redisvFormatCommand(&cmd, format, ap);
	if (len == -1) {
		__redisClusterSetError(cc, REDIS_ERR_OOM, "Out of memory");
		return REDIS_ERR;
	}
	else if (len == -2) {
		__redisClusterSetError(cc, REDIS_ERR_OTHER, "Invalid format string");
		return REDIS_ERR;
	}

	ret = redisClusterAppendFormattedCommand(cc, cmd, len);

	free(cmd);

	return ret;
}

int redisClusterAppendCommand(redisClusterContext *cc,
	const char *format, ...) {

	int ret;
	va_list ap;

	if (cc == NULL || format == NULL)
	{
		return REDIS_ERR;
	}

	va_start(ap, format);
	ret = redisClustervAppendCommand(cc, format, ap);
	va_end(ap);

	return ret;
}

int redisClusterAppendCommandArgv(redisClusterContext *cc,
	int argc, const char **argv, const size_t *argvlen) {
	int ret;
	char *cmd;
	int len;

	len = redisFormatCommandArgv(&cmd, argc, argv, argvlen);
	if (len == -1) {
		__redisClusterSetError(cc, REDIS_ERR_OOM, "Out of memory");
		return REDIS_ERR;
	}

	ret = redisClusterAppendFormattedCommand(cc, cmd, len);

	free(cmd);

	return ret;
}

static int redisCLusterSendAll(redisClusterContext *cc)
{
	dictIterator *di;
	dictEntry *de;
	struct cluster_node *node;
	redisContext *c = NULL;
	int wdone = 0;

	if (cc == NULL || cc->nodes == NULL)
	{
		return REDIS_ERR;
	}

	di = dictGetIterator(cc->nodes);
	while ((de = dictNext(di)) != NULL)
	{
		node = dictGetEntryVal(de);
		if (node == NULL)
		{
			continue;
		}

		c = ctx_get_by_node(cc, node);
		if (c == NULL)
		{
			continue;
		}

		if (c->flags & REDIS_BLOCK) {
			/* Write until done */
			do {
				if (redisBufferWrite(c, &wdone) == REDIS_ERR)
				{
					dictReleaseIterator(di);
					return REDIS_ERR;
				}
			} while (!wdone);
		}
	}

	dictReleaseIterator(di);

	return REDIS_OK;
}

static int redisCLusterClearAll(redisClusterContext *cc)
{
	dictIterator *di;
	dictEntry *de;
	struct cluster_node *node;
	redisContext *c = NULL;

	if (cc == NULL) {
		return REDIS_ERR;
	}

	if (cc->err) {
		cc->err = 0;
		// memset(cc->errstr, '\0', strlen(cc->errstr));
		cc->errstr[0] = '\0';
	}

	if (cc->nodes == NULL) {
		return REDIS_ERR;
	}
	di = dictGetIterator(cc->nodes);
	while ((de = dictNext(di)) != NULL)
	{
		node = dictGetEntryVal(de);
		if (node == NULL)
		{
			continue;
		}

		c = node->con;
		if (c == NULL)
		{
			continue;
		}

		redisFree(c);
		node->con = NULL;
	}

	dictReleaseIterator(di);

	return REDIS_OK;
}

int redisClusterGetReply(redisClusterContext *cc, void **reply) {

	struct cmd *command, *sub_command;
	hilist *commands = NULL;
	listNode *list_command, *list_sub_command;
	listIter *list_iter;
	int slot_num;
	void *sub_reply;

	if (cc == NULL || reply == NULL)
		return REDIS_ERR;

	cc->err = 0;
	cc->errstr[0] = '\0';

	*reply = NULL;

	if (cc->requests == NULL)
		return REDIS_ERR;

	list_command = listFirst(cc->requests);

	//no more reply
	if (list_command == NULL)
	{
		*reply = NULL;
		return REDIS_OK;
	}

	command = list_command->value;
	if (command == NULL)
	{
		__redisClusterSetError(cc, REDIS_ERR_OTHER,
			"command in the requests list is null");
		goto error;
	}

	slot_num = command->slot_num;
	if (slot_num >= 0)
	{
		listDelNode(cc->requests, list_command);
		return __redisClusterGetReply(cc, slot_num, reply);
	}

	commands = command->sub_commands;
	if (commands == NULL)
	{
		__redisClusterSetError(cc, REDIS_ERR_OTHER,
			"sub_commands in command is null");
		goto error;
	}

	ASSERT(listLength(commands) != 1);

	list_iter = listGetIterator(commands, AL_START_HEAD);
	while ((list_sub_command = listNext(list_iter)) != NULL)
	{
		sub_command = list_sub_command->value;
		if (sub_command == NULL)
		{
			__redisClusterSetError(cc, REDIS_ERR_OTHER,
				"sub_command is null");
			goto error;
		}

		slot_num = sub_command->slot_num;
		if (slot_num < 0)
		{
			__redisClusterSetError(cc, REDIS_ERR_OTHER,
				"sub_command slot_num is less then zero");
			goto error;
		}

		if (__redisClusterGetReply(cc, slot_num, &sub_reply) != REDIS_OK)
		{
			goto error;
		}

		sub_command->reply = sub_reply;
	}

	*reply = command_post_fragment(cc, command, commands);
	if (*reply == NULL)
	{
		goto error;
	}

	listDelNode(cc->requests, list_command);
	return REDIS_OK;

error:

	listDelNode(cc->requests, list_command);
	return REDIS_ERR;
}

void redisClusterReset(redisClusterContext *cc)
{
	int status;
	void *reply;

	if (cc == NULL || cc->nodes == NULL)
	{
		return;
	}

	if (cc->err) {
		redisCLusterClearAll(cc);
	}
	else {
		redisCLusterSendAll(cc);

		do {
			status = redisClusterGetReply(cc, &reply);
			if (status == REDIS_OK) {
				freeReplyObject(reply);
			}
			else {
				redisCLusterClearAll(cc);
				break;
			}
		} while (reply != NULL);
	}

	if (cc->requests)
	{
		listRelease(cc->requests);
		cc->requests = NULL;
	}

	if (cc->need_update_route)
	{
		status = cluster_update_route(cc);
		if (status != REDIS_OK)
		{
			__redisClusterSetError(cc, REDIS_ERR_OTHER,
				"route update error, please recreate redisClusterContext!");
			return;
		}
		cc->need_update_route = 0;
	}
}

/*############redis cluster async############*/

/* We want the error field to be accessible directly instead of requiring
 * an indirection to the redisContext struct. */
static void __redisClusterAsyncCopyError(redisClusterAsyncContext *acc) {
	if (!acc)
		return;

	redisClusterContext *cc = acc->cc;
	acc->err = cc->err;
	memcpy(acc->errstr, cc->errstr, 128);
}

static void __redisClusterAsyncSetError(redisClusterAsyncContext *acc,
	int type, const char *str) {

	acc->err = type;
	if (str != NULL) {
		strncpy(acc->errstr, str, sizeof(acc->errstr));
	}
	else {
		/* Only REDIS_ERR_IO may lack a description! */
		assert(type == REDIS_ERR_IO);
		__redis_strerror_r(errno, acc->errstr, sizeof(acc->errstr));
	}
}

static redisClusterAsyncContext *redisClusterAsyncInitialize(redisClusterContext *cc) {
	redisClusterAsyncContext *acc;

	if (cc == NULL)
	{
		return NULL;
	}

	acc = hi_alloc(sizeof(redisClusterAsyncContext));
	if (acc == NULL)
		return NULL;

	acc->cc = cc;

	acc->err = 0;
	acc->data = NULL;
	acc->adapter = NULL;
	acc->attach_fn = NULL;

	acc->onConnect = NULL;
	acc->onDisconnect = NULL;

	return acc;
}

static cluster_async_data *cluster_async_data_get(void)
{
	cluster_async_data *cad;

	cad = hi_alloc(sizeof(cluster_async_data));
	if (cad == NULL)
	{
		return NULL;
	}

	cad->acc = NULL;
	cad->command = NULL;
	cad->callback = NULL;
	cad->privdata = NULL;
	cad->retry_count = 0;

	return cad;
}

static void cluster_async_data_free(cluster_async_data *cad)
{
	if (cad == NULL)
	{
		return;
	}

	if (cad->command != NULL)
	{
		command_destroy(cad->command);
	}

	hi_free(cad);
	cad = NULL;
}

static void unlinkAsyncContextAndNode(redisAsyncContext* ac)
{
	cluster_node *node;

	if (ac->data) {
		node = (cluster_node *)(ac->data);
		node->acon = NULL;
	}
}

redisAsyncContext * actx_get_by_node(redisClusterAsyncContext *acc,
	cluster_node *node)
{
	redisAsyncContext *ac;

	if (node == NULL)
	{
		return NULL;
	}

	ac = node->acon;
	if (ac != NULL)
	{
		if (ac->c.err == 0) {
			return ac;
		}
		else {
			NOT_REACHED();
		}
	}

	if (node->host == NULL || node->port <= 0)
	{
		__redisClusterAsyncSetError(acc, REDIS_ERR_OTHER, "node host or port is error");
		return NULL;
	}

	ac = redisAsyncConnect(node->host, node->port);
	if (ac == NULL)
	{
		__redisClusterAsyncSetError(acc, REDIS_ERR_OTHER, "node host or port is error");
		return NULL;
	}

	if (acc->cc->password[0]) {
		redisAsyncCommand(ac, NULL, NULL, "AUTH %s", acc->cc->password);
		printf("auth after connect.\n");
	}

	// _redisClusterAuth(acc->cc, &ac->c);

	if (acc->adapter)
	{
		acc->attach_fn(ac, acc->adapter);
	}

	if (acc->onConnect)
	{
		redisAsyncSetConnectCallback(ac, acc->onConnect);
	}

	if (acc->onDisconnect)
	{
		redisAsyncSetDisconnectCallback(ac, acc->onDisconnect);
	}

	ac->data = node;
	ac->dataHandler = unlinkAsyncContextAndNode;
	node->acon = ac;

	return ac;
}

static redisAsyncContext *actx_get_after_update_route_by_slot(
	redisClusterAsyncContext *acc, int slot_num)
{
	int ret;
	redisClusterContext *cc;
	redisAsyncContext *ac;
	cluster_node *node;

	if (acc == NULL || slot_num < 0)
	{
		return NULL;
	}

	cc = acc->cc;
	if (cc == NULL)
	{
		return NULL;
	}

	ret = cluster_update_route(cc);
	if (ret != REDIS_OK)
	{
		__redisClusterAsyncSetError(acc, REDIS_ERR_OTHER,
			"route update error, please recreate redisClusterContext!");
		return NULL;
	}

	node = node_get_by_table(cc, (uint32_t)slot_num);
	if (node == NULL)
	{
		__redisClusterAsyncSetError(acc,
			REDIS_ERR_OTHER, "node get by table error");
		return NULL;
	}

	ac = actx_get_by_node(acc, node);
	if (ac == NULL)
	{
		__redisClusterAsyncSetError(acc,
			REDIS_ERR_OTHER, "actx get by node error");
		return NULL;
	}
	else if (ac->err)
	{
		__redisClusterAsyncSetError(acc, ac->err, ac->errstr);
		return NULL;
	}

	return ac;
}

redisClusterAsyncContext *redisClusterAsyncConnect(const char *addrs, const char *password, int flags) {

	redisClusterContext *cc;
	redisClusterAsyncContext *acc;

	cc = redisClusterConnectNonBlock(addrs, password, flags);
	if (cc == NULL)
	{
		return NULL;
	}

	acc = redisClusterAsyncInitialize(cc);
	if (acc == NULL) {
		redisClusterFree(cc);
		return NULL;
	}

	__redisClusterAsyncCopyError(acc);

	return acc;
}


int redisClusterAsyncSetConnectCallback(
	redisClusterAsyncContext *acc, redisConnectCallback *fn)
{
	if (acc->onConnect == NULL) {
		acc->onConnect = fn;
		return REDIS_OK;
	}
	return REDIS_ERR;
}

int redisClusterAsyncSetDisconnectCallback(
	redisClusterAsyncContext *acc, redisDisconnectCallback *fn)
{
	if (acc->onDisconnect == NULL) {
		acc->onDisconnect = fn;
		return REDIS_OK;
	}
	return REDIS_ERR;
}

static void redisClusterAsyncCallback(redisAsyncContext *ac, void *r, void *privdata) {
	int ret;
	redisReply *reply = r;
	cluster_async_data *cad = privdata;
	redisClusterAsyncContext *acc;
	redisClusterContext *cc;
	redisAsyncContext *ac_retry = NULL;
	int error_type;
	cluster_node *node;
	struct cmd *command;
	int64_t now, next;

	if (cad == NULL)
	{
		goto error;
	}

	acc = cad->acc;
	if (acc == NULL)
	{
		goto error;
	}

	cc = acc->cc;
	if (cc == NULL)
	{
		goto error;
	}

	command = cad->command;
	if (command == NULL)
	{
		goto error;
	}

	if (reply == NULL)
	{
		//Note: 
		//I can't decide witch is the best way to deal with connect 
		//problem for hiredis cluster async api.
		//But now the way is : when enough null reply for a node,
		//we will update the route after the cluster node timeout.
		//If you have a better idea, please contact with me. Thank you.
		//My email: diguo58@gmail.com

		node = (cluster_node *)(ac->data);
		ASSERT(node != NULL);

		__redisClusterAsyncSetError(acc,
			ac->err, ac->errstr);

		if (cc->update_route_time != 0)
		{
			now = hi_usec_now();
			if (now >= cc->update_route_time)
			{
				ret = cluster_update_route(cc);
				if (ret != REDIS_OK)
				{
					__redisClusterAsyncSetError(acc, REDIS_ERR_OTHER,
						"route update error, please recreate redisClusterContext!");
				}

				cc->update_route_time = 0LL;
			}

			goto done;
		}

		node->failure_count++;
		if (node->failure_count > cc->max_redirect_count)
		{
			char *cluster_timeout_str;
			int cluster_timeout_str_len;
			int cluster_timeout;

			node->failure_count = 0;
			if (cc->update_route_time != 0)
			{
				goto done;
			}

			cluster_timeout_str = cluster_config_get(cc,
				"cluster-node-timeout", &cluster_timeout_str_len);
			if (cluster_timeout_str == NULL)
			{
				__redisClusterAsyncSetError(acc,
					cc->err, cc->errstr);
				goto done;
			}

			cluster_timeout = hi_atoi(cluster_timeout_str,
				cluster_timeout_str_len);
			free(cluster_timeout_str);
			if (cluster_timeout <= 0)
			{
				__redisClusterAsyncSetError(acc,
					REDIS_ERR_OTHER,
					"cluster_timeout_str convert to integer error");
				goto done;
			}

			now = hi_usec_now();
			if (now < 0) {
				__redisClusterAsyncSetError(acc,
					REDIS_ERR_OTHER,
					"get now usec time error");
				goto done;
			}

			next = now + (cluster_timeout * 1000LL);

			cc->update_route_time = next;

		}

		goto done;
	}

	error_type = cluster_reply_error_type(reply);

	if (error_type > CLUSTER_NOT_ERR && error_type < CLUSTER_ERR_SENTINEL)
	{
		cad->retry_count++;
		if (cad->retry_count > cc->max_redirect_count)
		{
			cad->retry_count = 0;
			__redisClusterAsyncSetError(acc,
				REDIS_ERR_CLUSTER_TOO_MANY_REDIRECT,
				"too many cluster redirect");
			goto done;
		}

		switch (error_type)
		{
		case CLUSTER_ERR_MOVED:
			ac_retry = actx_get_after_update_route_by_slot(acc, command->slot_num);
			if (ac_retry == NULL)
			{
				goto done;
			}

			break;
		case CLUSTER_ERR_ASK:
			node = node_get_by_ask_error_reply(cc, reply);
			if (node == NULL)
			{
				__redisClusterAsyncSetError(acc,
					cc->err, cc->errstr);
				goto done;
			}

			ac_retry = actx_get_by_node(acc, node);
			if (ac_retry == NULL)
			{
				__redisClusterAsyncSetError(acc,
					REDIS_ERR_OTHER, "actx get by node error");
				goto done;
			}
			else if (ac_retry->err)
			{
				__redisClusterAsyncSetError(acc,
					ac_retry->err, ac_retry->errstr);
				goto done;
			}

			ret = redisAsyncCommand(ac_retry,
				NULL, NULL, REDIS_COMMAND_ASKING);
			if (ret != REDIS_OK)
			{
				goto error;
			}

			break;
		case CLUSTER_ERR_TRYAGAIN:
		case CLUSTER_ERR_CROSSSLOT:
		case CLUSTER_ERR_CLUSTERDOWN:
			ac_retry = ac;

			break;
		default:

			goto done;
			break;
		}

		goto retry;
	}

done:

	if (acc->err)
	{
		cad->callback(acc, NULL, cad->privdata);
	}
	else
	{
		cad->callback(acc, r, cad->privdata);
	}

	if (cc->err)
	{
		cc->err = 0;
		// memset(cc->errstr, '\0', strlen(cc->errstr));
		cc->errstr[0] = '\0';
	}

	if (acc->err)
	{
		acc->err = 0;
		acc->errstr[0] = '\0';
		// memset(acc->errstr, '\0', strlen(acc->errstr));
	}

	
	if (cad != NULL)
	{
		// if cmd is subscribe/psubscribe, don't free!
		if(cad->command->type != CMD_REQ_REDIS_SUBSCRIBE
			&& cad->command->type != CMD_REQ_REDIS_PSUBSCRIBE)
			cluster_async_data_free(cad);
	}
	

	return;

retry:

	ret = redisAsyncFormattedCommand(ac_retry,
		redisClusterAsyncCallback, cad, command->cmd, command->clen);
	if (ret != REDIS_OK)
	{
		goto error;
	}

	return;

error:

	if (cad != NULL)
	{
		cluster_async_data_free(cad);
	}
}

int redisClusterAsyncFormattedCommand(redisClusterAsyncContext *acc,
	redisClusterCallbackFn *fn, void *privdata, char *cmd, int len) {

	redisClusterContext *cc;
	int status = REDIS_OK;
	int slot_num;
	cluster_node *node;
	redisAsyncContext *ac;
	struct cmd *command = NULL;
	hilist *commands = NULL;
	cluster_async_data *cad;

	if (acc == NULL)
	{
		return REDIS_ERR;
	}

	cc = acc->cc;

	if (cc->err)
	{
		cc->err = 0;
		// memset(cc->errstr, '\0', strlen(cc->errstr));
		cc->errstr[0] = '\0';
	}

	if (acc->err)
	{
		acc->err = 0;
		// memset(acc->errstr, '\0', strlen(acc->errstr));
		acc->errstr[0] = '\0';
	}

	command = command_get();
	if (command == NULL)
	{
		__redisClusterAsyncSetError(acc, REDIS_ERR_OOM, "Out of memory");
		goto error;
	}

	command->cmd = malloc(len * sizeof(*command->cmd));
	if (command->cmd == NULL)
	{
		__redisClusterAsyncSetError(acc, REDIS_ERR_OOM, "Out of memory");
		goto error;
	}
	memcpy(command->cmd, cmd, len);
	command->clen = len;

	commands = listCreate();
	if (commands == NULL)
	{
		__redisClusterAsyncSetError(acc, REDIS_ERR_OOM, "Out of memory");
		goto error;
	}

	commands->free = listCommandFree;

	slot_num = command_format_by_slot(cc, command, commands);

	if (slot_num < 0)
	{
		__redisClusterAsyncSetError(acc,
			cc->err, cc->errstr);
		goto error;
	}
	else if (slot_num >= REDIS_CLUSTER_SLOTS)
	{
		__redisClusterAsyncSetError(acc,
			REDIS_ERR_OTHER, "slot_num is out of range");
		goto error;
	}

	//all keys not belong to one slot
	if (listLength(commands) > 0)
	{
		ASSERT(listLength(commands) != 1);

		__redisClusterAsyncSetError(acc, REDIS_ERR_OTHER,
			"Asynchronous API now not support multi-key command");
		goto error;
	}

	node = node_get_by_table(cc, (uint32_t)slot_num);
	if (node == NULL)
	{
		__redisClusterAsyncSetError(acc,
			REDIS_ERR_OTHER, "node get by table error");
		goto error;
	}

	ac = actx_get_by_node(acc, node);
	if (ac == NULL)
	{
		__redisClusterAsyncSetError(acc,
			REDIS_ERR_OTHER, "actx get by node error");
		goto error;
	}
	else if (ac->err)
	{
		__redisClusterAsyncSetError(acc, ac->err, ac->errstr);
		goto error;
	}

	cad = cluster_async_data_get();
	if (cad == NULL)
	{
		__redisClusterAsyncSetError(acc, REDIS_ERR_OOM, "Out of memory");
		goto error;
	}

	cad->acc = acc;
	cad->command = command;
	cad->callback = fn;
	cad->privdata = privdata;

	status = redisAsyncFormattedCommand(ac,
		redisClusterAsyncCallback, cad, cmd, len);
	if (status != REDIS_OK)
	{
		goto error;
	}

	if (commands != NULL)
	{
		listRelease(commands);
	}

	return REDIS_OK;

error:

	if (command != NULL)
	{
		command_destroy(command);
	}

	if (commands != NULL)
	{
		listRelease(commands);
	}

	return REDIS_ERR;
}


int redisClustervAsyncCommand(redisClusterAsyncContext *acc,
	redisClusterCallbackFn *fn, void *privdata, const char *format, va_list ap) {
	int ret;
	char *cmd;
	int len;

	if (acc == NULL)
	{
		return REDIS_ERR;
	}

	len = redisvFormatCommand(&cmd, format, ap);
	if (len == -1) {
		__redisClusterAsyncSetError(acc, REDIS_ERR_OOM, "Out of memory");
		return REDIS_ERR;
	}
	else if (len == -2) {
		__redisClusterAsyncSetError(acc, REDIS_ERR_OTHER, "Invalid format string");
		return REDIS_ERR;
	}

	ret = redisClusterAsyncFormattedCommand(acc, fn, privdata, cmd, len);

	free(cmd);

	return ret;
}

int redisClusterAsyncCommand(redisClusterAsyncContext *acc,
	redisClusterCallbackFn *fn, void *privdata, const char *format, ...) {
	int ret;
	va_list ap;

	va_start(ap, format);
	ret = redisClustervAsyncCommand(acc, fn, privdata, format, ap);
	va_end(ap);

	return ret;
}

int redisClusterAsyncCommandArgv(redisClusterAsyncContext *acc,
	redisClusterCallbackFn *fn, void *privdata, int argc, const char **argv, const size_t *argvlen) {
	int ret;
	char *cmd;
	int len;

	len = redisFormatCommandArgv(&cmd, argc, argv, argvlen);
	if (len == -1) {
		__redisClusterAsyncSetError(acc, REDIS_ERR_OOM, "Out of memory");
		return REDIS_ERR;
	}

	ret = redisClusterAsyncFormattedCommand(acc, fn, privdata, cmd, len);

	free(cmd);

	return ret;
}

void redisClusterAsyncDisconnect(redisClusterAsyncContext *acc) {

	redisClusterContext *cc;
	redisAsyncContext *ac;
	dictIterator *di;
	dictEntry *de;
	dict *nodes;
	struct cluster_node *node;

	if (acc == NULL)
	{
		return;
	}

	cc = acc->cc;

	nodes = cc->nodes;

	if (nodes == NULL)
	{
		return;
	}

	di = dictGetIterator(nodes);

	while ((de = dictNext(di)) != NULL)
	{
		node = dictGetEntryVal(de);

		ac = node->acon;

		if (ac == NULL || ac->err)
		{
			continue;
		}

		redisAsyncDisconnect(ac);

		node->acon = NULL;
	}
}

void redisClusterAsyncFree(redisClusterAsyncContext *acc)
{
	redisClusterContext *cc;

	if (acc == NULL)
	{
		return;
	}

	cc = acc->cc;

	redisClusterFree(cc);

	hi_free(acc);
}

